import asyncio
import aiohttp
from aiohttp import web
import subprocess
import shlex
async def handle_terminal(websocket, path):
prompt_active = False
prompt_response = None
try:
async for message in websocket:
command = shlex.split(message.data)
if not command or command[0] in ['rm', 'sudo', 'halt', 'reboot']:
await websocket.send_str("Command not allowed")
continue
try:
process = await asyncio.create_subprocess_exec(
*command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
async for line in process.stdout:
# Check if line contains a prompt
line_str = line.decode('utf-8').rstrip()
if line_str.endswith('[Y/n]') or line_str.endswith('[y/n]'):
await websocket.send_str(line_str)
prompt_active = True
# Wait for user response
while prompt_active:
prompt_response = await websocket.receive_str()
if prompt_response.lower() in ['y', 'n']:
prompt_active = False
await websocket.send_str("") # Signal end of prompt
else:
await websocket.send_str(line_str)
async for line in process.stderr:
await websocket.send_str(line.decode('utf-8').rstrip())
await process.wait()
await websocket.send_str("")
except FileNotFoundError:
await websocket.send_str(f"Command not found: {command[0]}")
except Exception as e:
await websocket.send_str(f"Error: {str(e)}")
except aiohttp.WSCloseCode:
pass
async def index(request):
return web.Response(
text="""
Web Terminal