kali-linux-server / server.py
phonglanvq003's picture
Upload server.py with huggingface_hub
1d73eca verified
"""
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")
if not SECRET_TOKEN:
import secrets
SECRET_TOKEN = secrets.token_urlsafe(32)
print(f"⚠️ No API_TOKEN found in env. Generated temporary token: {SECRET_TOKEN}")
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)