Spaces:

proti0070
/

kali

Paused

kali / server.py

Update server.py

a660820 verified 8 months ago

18.5 kB

"""
HTTP + WebSocket Remote Shell Server
SSH Replacement với FastAPI
"""
import asyncio
import os
import signal
import pty
import subprocess
import io
import zipfile
import struct
import fcntl
import termios
from pathlib import Path
from typing import Optional

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query, HTTPException, Depends, UploadFile, File as FastAPIFile
from fastapi.responses import StreamingResponse, FileResponse, HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI(title="Remote Shell")

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# =============================================================================
# Configuration
# =============================================================================
# Use a more secure default if no token provided
SECRET_TOKEN = os.getenv("API_TOKEN", "shell")

WORK_DIR = Path(os.getenv("WORK_DIR", "/tmp/remote-shell")).resolve()
WORK_DIR.mkdir(parents=True, exist_ok=True)

security = HTTPBearer(auto_error=False)

# Store running processes
running_processes: dict = {}
pty_sessions: dict = {}


# =============================================================================
# Auth Middleware
# =============================================================================
async def verify_token(credentials: Optional[HTTPAuthorizationCredentials] = Depends(security)):
    """Verify Bearer token"""
    if credentials is None or credentials.credentials != SECRET_TOKEN:
        raise HTTPException(status_code=401, detail="Invalid or missing token")
    return credentials.credentials


def get_safe_path(path: str, session_id: str = None) -> Path:
    """Resolve path, supporting ~ for home directory. If session_id is provided, resolve relative to session workspace."""
    # Base directory depends on session_id
    base = WORK_DIR
    if session_id:
        # Sanitize session_id to prevent ".." in session_id
        safe_session_id = "".join(c for c in session_id if c.isalnum() or c in ("-", "_"))
        base = (WORK_DIR / safe_session_id).resolve()
        base.mkdir(parents=True, exist_ok=True)
    
    # Expand ~ to home directory (or base dir if in session)
    if path.startswith('~'):
        if session_id:
            path = path.replace('~', '.', 1)
        else:
            path = os.path.expanduser(path)
    
    # Resolve the path relative to base
    # This correctly handles both absolute paths (bypassing base if not careful) 
    # and relative paths.
    requested_path = Path(path)
    
    if requested_path.is_absolute():
        # If the server is in 'Global' mode (no session), allow absolute paths
        # If in 'Session' mode, absolute paths MUST be within the session base
        full_path = requested_path.resolve()
    else:
        full_path = (base / path).resolve()
    
    # CRITICAL: Path Traversal Check
    # Ensure the resolved path is still inside the intended base
    if not str(full_path).startswith(str(base)):
        raise HTTPException(status_code=403, detail=f"Access denied: path {path} is outside the allowed workspace")
    
    return full_path


# =============================================================================
# Command Execution (HTTP Streaming)
# =============================================================================
async def run_command_generator(cmd: str, session_id: str):
    """Generator to stream command output"""
    process = await asyncio.create_subprocess_shell(
        cmd,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.STDOUT,
        cwd=str(WORK_DIR),
        preexec_fn=os.setsid
    )
    running_processes[session_id] = process

    try:
        while True:
            line = await process.stdout.readline()
            if not line:
                break
            yield line.decode('utf-8', errors='replace')
    finally:
        await process.wait()
        running_processes.pop(session_id, None)


@app.get("/exec")
async def stream_exec(
    cmd: str,
    session_id: str = Query(default="default"),
    token: str = Depends(verify_token)
):
    """Execute command with streaming output"""
    return StreamingResponse(
        run_command_generator(cmd, session_id),
        media_type="text/plain"
    )


# =============================================================================
# File Operations
# =============================================================================
@app.get("/files")
async def list_files(
    path: str = Query(default="."),
    show_hidden: bool = Query(default=True),
    session_id: str = Query(default=None),
    token: str = Depends(verify_token)
):
    """List directory contents including hidden files"""
    target = get_safe_path(path, session_id)
    if not target.exists():
        raise HTTPException(404, "Path not found")
    
    if target.is_file():
        stat = target.stat()
        return [{
            "name": target.name,
            "type": "file",
            "size": stat.st_size
        }]
    
    items = []
    for item in target.iterdir():
        # Skip hidden files if show_hidden is False
        if not show_hidden and item.name.startswith('.'):
            continue
        try:
            stat = item.stat()
            items.append({
                "name": item.name,
                "type": "directory" if item.is_dir() else "file",
                "size": stat.st_size if item.is_file() else None,
                "hidden": item.name.startswith('.')
            })
        except (PermissionError, OSError):
            # Skip files we can't access
            continue
    
    # Sort: directories first, then by name
    return sorted(items, key=lambda x: (x["type"] == "file", x["name"].lower()))


@app.get("/files/read")
async def read_file(
    path: str,
    session_id: str = Query(default=None),
    token: str = Depends(verify_token)
):
    """Read file content"""
    target = get_safe_path(path, session_id)
    if not target.exists():
        raise HTTPException(404, "File not found")
    if target.is_dir():
        raise HTTPException(400, "Cannot read directory")
    
    try:
        content = target.read_text(errors='replace')
        return {"path": path, "content": content}
    except Exception as e:
        raise HTTPException(500, str(e))


@app.post("/files/write")
async def write_file(
    path: str,
    content: str,
    session_id: str = Query(default=None),
    token: str = Depends(verify_token)
):
    """Write content to file"""
    target = get_safe_path(path, session_id)
    target.parent.mkdir(parents=True, exist_ok=True)
    
    try:
        target.write_text(content)
        return {"status": "ok", "path": path, "size": len(content)}
    except Exception as e:
        raise HTTPException(500, str(e))


@app.post("/files/upload-zip")
async def upload_zip_form(
    file: UploadFile = FastAPIFile(...),
    path: str = Query(default="."),
    session_id: str = Query(default=None),
    token: str = Depends(verify_token)
):
    """Upload and extract a zip file (multipart form).
    
    Args:
        file: Uploaded zip file
        path: Directory to extract to
        session_id: Session identifier for workspace isolation
    
    Returns:
        folder_path, tree
    """
    try:
        # Read uploaded file
        zip_bytes = await file.read()
        zip_filename = file.filename or "upload.zip"
        
        # Determine extract location
        target_dir = get_safe_path(path, session_id)
        target_dir.mkdir(parents=True, exist_ok=True)
        
        # Remove .zip extension for folder name
        folder_name = zip_filename.rsplit('.', 1)[0] if '.' in zip_filename else zip_filename
        folder_name = folder_name.replace(' ', '_').replace('/', '_')
        
        extract_dir = target_dir / folder_name
        extract_dir.mkdir(parents=True, exist_ok=True)
        
        # Extract zip
        try:
            with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
                zf.extractall(extract_dir)
        except zipfile.BadZipFile:
            raise HTTPException(400, "Invalid zip file")
        
        # Get folder tree (limit depth to 3)
        tree_items = []
        for item in extract_dir.rglob("*"):
            rel = item.relative_to(extract_dir)
            if len(rel.parts) <= 3:
                tree_items.append(str(rel))
            if len(tree_items) >= 50:
                break
        
        return {
            "folder_path": str(extract_dir),
            "tree": "\n".join(tree_items)
        }
        
    except HTTPException:
        raise
    except Exception as e:
        return {"success": False, "error": str(e)}


# =============================================================================
# WebSocket PTY Terminal
# =============================================================================
@app.websocket("/ws/terminal")
async def websocket_terminal(
    websocket: WebSocket, 
    token: str = Query(...),
    work_dir: str = Query(default=None),  # None = use server user's home
    session_id: str = Query(default="default")  # Session identifier for cwd tracking
):
    """WebSocket PTY - Full interactive terminal with custom work directory"""
    # Verify token
    if token != SECRET_TOKEN:
        await websocket.close(code=4001, reason="Unauthorized")
        return
    
    # Default to home folder of the user running the server
    if work_dir is None or work_dir == "" or work_dir == "~":
        work_dir = os.path.expanduser("~")
    
    # Validate and create work directory (expand ~ to home)
    work_path = Path(os.path.expanduser(work_dir)).resolve()
    work_path.mkdir(parents=True, exist_ok=True)
    
    await websocket.accept()
    
    # Create PTY
    master_fd, slave_fd = pty.openpty()
    
    # Set initial terminal size
    winsize = struct.pack("HHHH", 24, 80, 0, 0)
    fcntl.ioctl(slave_fd, termios.TIOCSWINSZ, winsize)
    
    # Environment for the shell
    env = os.environ.copy()
    env.update({
        "TERM": "xterm-256color",
        "COLORTERM": "truecolor",
        "LANG": "en_US.UTF-8",
        "LC_ALL": "en_US.UTF-8",
        "HOME": str(work_path),
        "PWD": str(work_path),
        "SHELL": "/bin/bash",
    })
    
    # Spawn bash as login shell
    pid = os.fork()
    if pid == 0:
        # Child process
        os.setsid()
        os.dup2(slave_fd, 0)
        os.dup2(slave_fd, 1)
        os.dup2(slave_fd, 2)
        os.close(master_fd)
        os.close(slave_fd)
        os.chdir(str(work_path))
        os.execvpe("bash", ["bash", "--login"], env)
    
    os.close(slave_fd)
    
    # Store session for cwd tracking
    pty_sessions[session_id] = {"pid": pid, "work_dir": str(work_path)}
    
    # Set non-blocking
    flags = fcntl.fcntl(master_fd, fcntl.F_GETFL)
    fcntl.fcntl(master_fd, fcntl.F_SETFL, flags | os.O_NONBLOCK)
    
    async def read_pty():
        """Read from PTY and send to WebSocket"""
        loop = asyncio.get_event_loop()
        while True:
            try:
                await asyncio.sleep(0.01)
                try:
                    data = os.read(master_fd, 4096)
                    if data:
                        await websocket.send_bytes(data)
                except BlockingIOError:
                    pass
                except OSError:
                    break
            except Exception:
                break
    
    async def write_pty():
        """Read from WebSocket and write to PTY"""
        while True:
            try:
                data = await websocket.receive()
                if data["type"] == "websocket.receive":
                    if "bytes" in data:
                        os.write(master_fd, data["bytes"])
                    elif "text" in data:
                        text = data["text"]
                        # Handle resize command
                        if text.startswith("RESIZE:"):
                            try:
                                _, cols, rows = text.split(":")
                                winsize = struct.pack("HHHH", int(rows), int(cols), 0, 0)
                                fcntl.ioctl(master_fd, termios.TIOCSWINSZ, winsize)
                            except:
                                pass
                        else:
                            os.write(master_fd, text.encode())
            except WebSocketDisconnect:
                break
            except Exception:
                break
    
    try:
        # Run both tasks
        read_task = asyncio.create_task(read_pty())
        write_task = asyncio.create_task(write_pty())
        
        done, pending = await asyncio.wait(
            [read_task, write_task],
            return_when=asyncio.FIRST_COMPLETED
        )
        
        for task in pending:
            task.cancel()
    finally:
        os.close(master_fd)
        pty_sessions.pop(session_id, None)
        try:
            os.kill(pid, signal.SIGTERM)
            os.waitpid(pid, 0)
        except:
            pass


@app.get("/cwd")
async def get_cwd(
    session_id: str = Query(default="default"),
    token: str = Depends(verify_token)
):
    """Get current working directory of a terminal session"""
    session = pty_sessions.get(session_id)
    if not session:
        raise HTTPException(404, "Session not found")
    
    pid = session["pid"]
    try:
        # Read the symlink /proc/{pid}/cwd to get actual cwd
        cwd = os.readlink(f"/proc/{pid}/cwd")
        return {"cwd": cwd, "session_id": session_id}
    except (OSError, FileNotFoundError):
        # Fallback to initial work_dir
        return {"cwd": session["work_dir"], "session_id": session_id}


def get_process_info(pid: int) -> Optional[dict]:
    """Read process info from /proc/{pid}"""
    try:
        proc_path = Path(f"/proc/{pid}")
        if not proc_path.exists():
            return None
        
        # Read stat file for basic info
        stat_content = (proc_path / "stat").read_text()
        stat_parts = stat_content.split()
        
        # Parse comm (process name) - it's in parentheses
        comm_start = stat_content.find('(')
        comm_end = stat_content.rfind(')')
        name = stat_content[comm_start+1:comm_end] if comm_start >= 0 and comm_end >= 0 else "unknown"
        
        # After comm, the rest of stat fields
        after_comm = stat_content[comm_end+2:].split()
        state = after_comm[0] if len(after_comm) > 0 else "?"
        ppid = int(after_comm[1]) if len(after_comm) > 1 else 0
        
        # Read cmdline for full command
        try:
            cmdline = (proc_path / "cmdline").read_text().replace('\x00', ' ').strip()
        except:
            cmdline = name
        
        # Read status for memory info
        rss_kb = 0
        try:
            status_content = (proc_path / "status").read_text()
            for line in status_content.split('\n'):
                if line.startswith("VmRSS:"):
                    rss_kb = int(line.split()[1])
                    break
        except:
            pass
        
        return {
            "pid": pid,
            "ppid": ppid,
            "name": name,
            "cmd": cmdline or name,
            "state": state,
            "memory_kb": rss_kb,
            "children": []
        }
    except (PermissionError, FileNotFoundError, OSError, ValueError, IndexError):
        return None


def build_process_tree(root_pid: Optional[int] = None) -> list:
    """Build hierarchical process tree"""
    processes = {}
    
    # Read all processes from /proc
    proc_path = Path("/proc")
    for entry in proc_path.iterdir():
        if entry.name.isdigit():
            pid = int(entry.name)
            info = get_process_info(pid)
            if info:
                processes[pid] = info
    
    # Build tree structure
    for pid, proc in processes.items():
        ppid = proc["ppid"]
        if ppid in processes and ppid != pid:
            processes[ppid]["children"].append(proc)
    
    # If root_pid specified, return subtree; otherwise return root processes
    if root_pid and root_pid in processes:
        return [processes[root_pid]]
    
    # Return only root processes (ppid=0 or ppid not in our list)
    roots = []
    for pid, proc in processes.items():
        if proc["ppid"] == 0 or proc["ppid"] not in processes:
            roots.append(proc)
    
    # Sort by PID
    roots.sort(key=lambda x: x["pid"])
    return roots


@app.get("/processes")
async def get_processes(
    session_id: Optional[str] = Query(default=None),
    token: str = Depends(verify_token)
):
    """Get process tree, optionally filtered by session's shell PID"""
    root_pid = None
    
    if session_id:
        session = pty_sessions.get(session_id)
        if session:
            root_pid = session["pid"]
    
    tree = build_process_tree(root_pid)
    return {"processes": tree, "session_id": session_id}


@app.post("/processes/kill")
async def kill_process_by_pid(
    pid: int = Query(...),
    sig: int = Query(default=15),  # SIGTERM by default
    token: str = Depends(verify_token)
):
    """Kill a process by PID"""
    try:
        os.kill(pid, sig)
        return {"status": "ok", "pid": pid, "signal": sig}
    except ProcessLookupError:
        raise HTTPException(404, f"Process {pid} not found")
    except PermissionError:
        raise HTTPException(403, f"Permission denied to kill process {pid}")
    except Exception as e:
        raise HTTPException(500, str(e))


# =============================================================================
# Static Files & Index
# =============================================================================
static_dir = Path(__file__).parent / "static"


@app.get("/health")
async def health(token: str = Depends(verify_token)):
    return {"status": "ok", "work_dir": str(WORK_DIR)}


# Mount static files (at the end so API routes take precedence)
if static_dir.exists():
    app.mount("/", StaticFiles(directory=str(static_dir), html=True), name="static")


if __name__ == "__main__":
    import uvicorn
    # Hugging Face Spaces uses port 7860 by default
    port = int(os.getenv("PORT", 7860))
    uvicorn.run(app, host="0.0.0.0", port=port)

	"""
	HTTP + WebSocket Remote Shell Server
	SSH Replacement với FastAPI
	"""
	import asyncio
	import os
	import signal
	import pty
	import subprocess
	import io
	import zipfile
	import struct
	import fcntl
	import termios
	from pathlib import Path
	from typing import Optional

	from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query, HTTPException, Depends, UploadFile, File as FastAPIFile
	from fastapi.responses import StreamingResponse, FileResponse, HTMLResponse
	from fastapi.staticfiles import StaticFiles
	from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
	from fastapi.middleware.cors import CORSMiddleware

	app = FastAPI(title="Remote Shell")

	# Add CORS middleware
	app.add_middleware(
	CORSMiddleware,
	allow_origins=["*"],
	allow_credentials=True,
	allow_methods=["*"],
	allow_headers=["*"],
	)

	# =============================================================================
	# Configuration
	# =============================================================================
	# Use a more secure default if no token provided
	SECRET_TOKEN = os.getenv("API_TOKEN", "shell")

	WORK_DIR = Path(os.getenv("WORK_DIR", "/tmp/remote-shell")).resolve()
	WORK_DIR.mkdir(parents=True, exist_ok=True)

	security = HTTPBearer(auto_error=False)

	# Store running processes
	running_processes: dict = {}
	pty_sessions: dict = {}


	# =============================================================================
	# Auth Middleware
	# =============================================================================
	async def verify_token(credentials: Optional[HTTPAuthorizationCredentials] = Depends(security)):
	"""Verify Bearer token"""
	if credentials is None or credentials.credentials != SECRET_TOKEN:
	raise HTTPException(status_code=401, detail="Invalid or missing token")
	return credentials.credentials


	def get_safe_path(path: str, session_id: str = None) -> Path:
	"""Resolve path, supporting ~ for home directory. If session_id is provided, resolve relative to session workspace."""
	# Base directory depends on session_id
	base = WORK_DIR
	if session_id:
	# Sanitize session_id to prevent ".." in session_id
	safe_session_id = "".join(c for c in session_id if c.isalnum() or c in ("-", "_"))
	base = (WORK_DIR / safe_session_id).resolve()
	base.mkdir(parents=True, exist_ok=True)

	# Expand ~ to home directory (or base dir if in session)
	if path.startswith('~'):
	if session_id:
	path = path.replace('~', '.', 1)
	else:
	path = os.path.expanduser(path)

	# Resolve the path relative to base
	# This correctly handles both absolute paths (bypassing base if not careful)
	# and relative paths.
	requested_path = Path(path)

	if requested_path.is_absolute():
	# If the server is in 'Global' mode (no session), allow absolute paths
	# If in 'Session' mode, absolute paths MUST be within the session base
	full_path = requested_path.resolve()
	else:
	full_path = (base / path).resolve()

	# CRITICAL: Path Traversal Check
	# Ensure the resolved path is still inside the intended base
	if not str(full_path).startswith(str(base)):
	raise HTTPException(status_code=403, detail=f"Access denied: path {path} is outside the allowed workspace")

	return full_path


	# =============================================================================
	# Command Execution (HTTP Streaming)
	# =============================================================================
	async def run_command_generator(cmd: str, session_id: str):
	"""Generator to stream command output"""
	process = await asyncio.create_subprocess_shell(
	cmd,
	stdout=asyncio.subprocess.PIPE,
	stderr=asyncio.subprocess.STDOUT,
	cwd=str(WORK_DIR),
	preexec_fn=os.setsid
	)
	running_processes[session_id] = process

	try:
	while True:
	line = await process.stdout.readline()
	if not line:
	break
	yield line.decode('utf-8', errors='replace')
	finally:
	await process.wait()
	running_processes.pop(session_id, None)


	@app.get("/exec")
	async def stream_exec(
	cmd: str,
	session_id: str = Query(default="default"),
	token: str = Depends(verify_token)
	):
	"""Execute command with streaming output"""
	return StreamingResponse(
	run_command_generator(cmd, session_id),
	media_type="text/plain"
	)


	# =============================================================================
	# File Operations
	# =============================================================================
	@app.get("/files")
	async def list_files(
	path: str = Query(default="."),
	show_hidden: bool = Query(default=True),
	session_id: str = Query(default=None),
	token: str = Depends(verify_token)
	):
	"""List directory contents including hidden files"""
	target = get_safe_path(path, session_id)
	if not target.exists():
	raise HTTPException(404, "Path not found")

	if target.is_file():
	stat = target.stat()
	return [{
	"name": target.name,
	"type": "file",
	"size": stat.st_size
	}]

	items = []
	for item in target.iterdir():
	# Skip hidden files if show_hidden is False
	if not show_hidden and item.name.startswith('.'):
	continue
	try:
	stat = item.stat()
	items.append({
	"name": item.name,
	"type": "directory" if item.is_dir() else "file",
	"size": stat.st_size if item.is_file() else None,
	"hidden": item.name.startswith('.')
	})
	except (PermissionError, OSError):
	# Skip files we can't access
	continue

	# Sort: directories first, then by name
	return sorted(items, key=lambda x: (x["type"] == "file", x["name"].lower()))


	@app.get("/files/read")
	async def read_file(
	path: str,
	session_id: str = Query(default=None),
	token: str = Depends(verify_token)
	):
	"""Read file content"""
	target = get_safe_path(path, session_id)
	if not target.exists():
	raise HTTPException(404, "File not found")
	if target.is_dir():
	raise HTTPException(400, "Cannot read directory")

	try:
	content = target.read_text(errors='replace')
	return {"path": path, "content": content}
	except Exception as e:
	raise HTTPException(500, str(e))


	@app.post("/files/write")
	async def write_file(
	path: str,
	content: str,
	session_id: str = Query(default=None),
	token: str = Depends(verify_token)
	):
	"""Write content to file"""
	target = get_safe_path(path, session_id)
	target.parent.mkdir(parents=True, exist_ok=True)

	try:
	target.write_text(content)
	return {"status": "ok", "path": path, "size": len(content)}
	except Exception as e:
	raise HTTPException(500, str(e))


	@app.post("/files/upload-zip")
	async def upload_zip_form(
	file: UploadFile = FastAPIFile(...),
	path: str = Query(default="."),
	session_id: str = Query(default=None),
	token: str = Depends(verify_token)
	):
	"""Upload and extract a zip file (multipart form).

	Args:
	file: Uploaded zip file
	path: Directory to extract to
	session_id: Session identifier for workspace isolation

	Returns:
	folder_path, tree
	"""
	try:
	# Read uploaded file
	zip_bytes = await file.read()
	zip_filename = file.filename or "upload.zip"

	# Determine extract location
	target_dir = get_safe_path(path, session_id)
	target_dir.mkdir(parents=True, exist_ok=True)

	# Remove .zip extension for folder name
	folder_name = zip_filename.rsplit('.', 1)[0] if '.' in zip_filename else zip_filename
	folder_name = folder_name.replace(' ', '_').replace('/', '_')

	extract_dir = target_dir / folder_name
	extract_dir.mkdir(parents=True, exist_ok=True)

	# Extract zip
	try:
	with zipfile.ZipFile(io.BytesIO(zip_bytes)) as zf:
	zf.extractall(extract_dir)
	except zipfile.BadZipFile:
	raise HTTPException(400, "Invalid zip file")

	# Get folder tree (limit depth to 3)
	tree_items = []
	for item in extract_dir.rglob("*"):
	rel = item.relative_to(extract_dir)
	if len(rel.parts) <= 3:
	tree_items.append(str(rel))
	if len(tree_items) >= 50:
	break

	return {
	"folder_path": str(extract_dir),
	"tree": "\n".join(tree_items)
	}

	except HTTPException:
	raise
	except Exception as e:
	return {"success": False, "error": str(e)}


	# =============================================================================
	# WebSocket PTY Terminal
	# =============================================================================
	@app.websocket("/ws/terminal")
	async def websocket_terminal(
	websocket: WebSocket,
	token: str = Query(...),
	work_dir: str = Query(default=None), # None = use server user's home
	session_id: str = Query(default="default") # Session identifier for cwd tracking
	):
	"""WebSocket PTY - Full interactive terminal with custom work directory"""
	# Verify token
	if token != SECRET_TOKEN:
	await websocket.close(code=4001, reason="Unauthorized")
	return

	# Default to home folder of the user running the server
	if work_dir is None or work_dir == "" or work_dir == "~":
	work_dir = os.path.expanduser("~")

	# Validate and create work directory (expand ~ to home)
	work_path = Path(os.path.expanduser(work_dir)).resolve()
	work_path.mkdir(parents=True, exist_ok=True)

	await websocket.accept()

	# Create PTY
	master_fd, slave_fd = pty.openpty()

	# Set initial terminal size
	winsize = struct.pack("HHHH", 24, 80, 0, 0)
	fcntl.ioctl(slave_fd, termios.TIOCSWINSZ, winsize)

	# Environment for the shell
	env = os.environ.copy()
	env.update({
	"TERM": "xterm-256color",
	"COLORTERM": "truecolor",
	"LANG": "en_US.UTF-8",
	"LC_ALL": "en_US.UTF-8",
	"HOME": str(work_path),
	"PWD": str(work_path),
	"SHELL": "/bin/bash",
	})

	# Spawn bash as login shell
	pid = os.fork()
	if pid == 0:
	# Child process
	os.setsid()
	os.dup2(slave_fd, 0)
	os.dup2(slave_fd, 1)
	os.dup2(slave_fd, 2)
	os.close(master_fd)
	os.close(slave_fd)
	os.chdir(str(work_path))
	os.execvpe("bash", ["bash", "--login"], env)

	os.close(slave_fd)

	# Store session for cwd tracking
	pty_sessions[session_id] = {"pid": pid, "work_dir": str(work_path)}

	# Set non-blocking
	flags = fcntl.fcntl(master_fd, fcntl.F_GETFL)
	fcntl.fcntl(master_fd, fcntl.F_SETFL, flags \| os.O_NONBLOCK)

	async def read_pty():
	"""Read from PTY and send to WebSocket"""
	loop = asyncio.get_event_loop()
	while True:
	try:
	await asyncio.sleep(0.01)
	try:
	data = os.read(master_fd, 4096)
	if data:
	await websocket.send_bytes(data)
	except BlockingIOError:
	pass
	except OSError:
	break
	except Exception:
	break

	async def write_pty():
	"""Read from WebSocket and write to PTY"""
	while True:
	try:
	data = await websocket.receive()
	if data["type"] == "websocket.receive":
	if "bytes" in data:
	os.write(master_fd, data["bytes"])
	elif "text" in data:
	text = data["text"]
	# Handle resize command
	if text.startswith("RESIZE:"):
	try:
	_, cols, rows = text.split(":")
	winsize = struct.pack("HHHH", int(rows), int(cols), 0, 0)
	fcntl.ioctl(master_fd, termios.TIOCSWINSZ, winsize)
	except:
	pass
	else:
	os.write(master_fd, text.encode())
	except WebSocketDisconnect:
	break
	except Exception:
	break

	try:
	# Run both tasks
	read_task = asyncio.create_task(read_pty())
	write_task = asyncio.create_task(write_pty())

	done, pending = await asyncio.wait(
	[read_task, write_task],
	return_when=asyncio.FIRST_COMPLETED
	)

	for task in pending:
	task.cancel()
	finally:
	os.close(master_fd)
	pty_sessions.pop(session_id, None)
	try:
	os.kill(pid, signal.SIGTERM)
	os.waitpid(pid, 0)
	except:
	pass


	@app.get("/cwd")
	async def get_cwd(
	session_id: str = Query(default="default"),
	token: str = Depends(verify_token)
	):
	"""Get current working directory of a terminal session"""
	session = pty_sessions.get(session_id)
	if not session:
	raise HTTPException(404, "Session not found")

	pid = session["pid"]
	try:
	# Read the symlink /proc/{pid}/cwd to get actual cwd
	cwd = os.readlink(f"/proc/{pid}/cwd")
	return {"cwd": cwd, "session_id": session_id}
	except (OSError, FileNotFoundError):
	# Fallback to initial work_dir
	return {"cwd": session["work_dir"], "session_id": session_id}


	def get_process_info(pid: int) -> Optional[dict]:
	"""Read process info from /proc/{pid}"""
	try:
	proc_path = Path(f"/proc/{pid}")
	if not proc_path.exists():
	return None

	# Read stat file for basic info
	stat_content = (proc_path / "stat").read_text()
	stat_parts = stat_content.split()

	# Parse comm (process name) - it's in parentheses
	comm_start = stat_content.find('(')
	comm_end = stat_content.rfind(')')
	name = stat_content[comm_start+1:comm_end] if comm_start >= 0 and comm_end >= 0 else "unknown"

	# After comm, the rest of stat fields
	after_comm = stat_content[comm_end+2:].split()
	state = after_comm[0] if len(after_comm) > 0 else "?"
	ppid = int(after_comm[1]) if len(after_comm) > 1 else 0

	# Read cmdline for full command
	try:
	cmdline = (proc_path / "cmdline").read_text().replace('\x00', ' ').strip()
	except:
	cmdline = name

	# Read status for memory info
	rss_kb = 0
	try:
	status_content = (proc_path / "status").read_text()
	for line in status_content.split('\n'):
	if line.startswith("VmRSS:"):
	rss_kb = int(line.split()[1])
	break
	except:
	pass

	return {
	"pid": pid,
	"ppid": ppid,
	"name": name,
	"cmd": cmdline or name,
	"state": state,
	"memory_kb": rss_kb,
	"children": []
	}
	except (PermissionError, FileNotFoundError, OSError, ValueError, IndexError):
	return None


	def build_process_tree(root_pid: Optional[int] = None) -> list:
	"""Build hierarchical process tree"""
	processes = {}

	# Read all processes from /proc
	proc_path = Path("/proc")
	for entry in proc_path.iterdir():
	if entry.name.isdigit():
	pid = int(entry.name)
	info = get_process_info(pid)
	if info:
	processes[pid] = info

	# Build tree structure
	for pid, proc in processes.items():
	ppid = proc["ppid"]
	if ppid in processes and ppid != pid:
	processes[ppid]["children"].append(proc)

	# If root_pid specified, return subtree; otherwise return root processes
	if root_pid and root_pid in processes:
	return [processes[root_pid]]

	# Return only root processes (ppid=0 or ppid not in our list)
	roots = []
	for pid, proc in processes.items():
	if proc["ppid"] == 0 or proc["ppid"] not in processes:
	roots.append(proc)

	# Sort by PID
	roots.sort(key=lambda x: x["pid"])
	return roots


	@app.get("/processes")
	async def get_processes(
	session_id: Optional[str] = Query(default=None),
	token: str = Depends(verify_token)
	):
	"""Get process tree, optionally filtered by session's shell PID"""
	root_pid = None

	if session_id:
	session = pty_sessions.get(session_id)
	if session:
	root_pid = session["pid"]

	tree = build_process_tree(root_pid)
	return {"processes": tree, "session_id": session_id}


	@app.post("/processes/kill")
	async def kill_process_by_pid(
	pid: int = Query(...),
	sig: int = Query(default=15), # SIGTERM by default
	token: str = Depends(verify_token)
	):
	"""Kill a process by PID"""
	try:
	os.kill(pid, sig)
	return {"status": "ok", "pid": pid, "signal": sig}
	except ProcessLookupError:
	raise HTTPException(404, f"Process {pid} not found")
	except PermissionError:
	raise HTTPException(403, f"Permission denied to kill process {pid}")
	except Exception as e:
	raise HTTPException(500, str(e))


	# =============================================================================
	# Static Files & Index
	# =============================================================================
	static_dir = Path(__file__).parent / "static"


	@app.get("/health")
	async def health(token: str = Depends(verify_token)):
	return {"status": "ok", "work_dir": str(WORK_DIR)}


	# Mount static files (at the end so API routes take precedence)
	if static_dir.exists():
	app.mount("/", StaticFiles(directory=str(static_dir), html=True), name="static")


	if __name__ == "__main__":
	import uvicorn
	# Hugging Face Spaces uses port 7860 by default
	port = int(os.getenv("PORT", 7860))
	uvicorn.run(app, host="0.0.0.0", port=port)