File size: 3,656 Bytes
6118034 15b7e8d 6118034 20555d8 df527c9 6118034 99d9fe9 df527c9 6118034 df527c9 8b4f8a5 df527c9 20555d8 6118034 20555d8 6118034 20555d8 6118034 20555d8 15b7e8d 6118034 20555d8 6118034 20555d8 6118034 20555d8 6118034 20555d8 6118034 20555d8 15b7e8d 20555d8 15b7e8d 20555d8 15b7e8d 20555d8 15b7e8d 20555d8 6118034 20555d8 6118034 20555d8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | import os
import asyncio
import json
import pty
import struct
import fcntl
import termios
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query
from fastapi.middleware.cors import CORSMiddleware
app = FastAPI()
PASSWORD = os.getenv("TERMINAL_PASSWORD")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, password: str = Query(default="")):
if password != PASSWORD:
await websocket.close(code=1008)
print("Conexão recusada: senha incorreta")
return
await websocket.accept()
pid, fd = pty.fork()
if pid == 0:
# Processo filho
os.environ["TERM"] = "xterm-256color"
shell = os.environ.get("SHELL", "/bin/bash")
os.execvp(shell, [shell])
else:
# Processo pai
loop = asyncio.get_event_loop()
async def read_from_terminal():
while True:
try:
data = await loop.run_in_executor(None, os.read, fd, 1024)
if data:
await websocket.send_text(data.decode(errors="ignore"))
else:
break
except Exception as e:
print("Erro ao ler do terminal:", e)
break
async def write_to_terminal():
try:
while True:
msg = await websocket.receive_text()
# Tenta parsear como JSON
try:
data = json.loads(msg)
# Handle resize
if data.get("type") == "resize":
cols = int(data.get("cols", 80))
rows = int(data.get("rows", 24))
# Método mais seguro usando ioctl
try:
winsize = struct.pack("HHHH", rows, cols, 0, 0)
fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize)
print(f"Terminal redimensionado: {cols}x{rows}")
except Exception as e:
print(f"Erro ao redimensionar terminal: {e}")
# Não envia nada para o terminal, apenas continua
continue
except json.JSONDecodeError:
# Se não for JSON, é input normal do usuário
os.write(fd, msg.encode())
except ValueError:
# Se houver erro nos valores do resize
print(f"Valores inválidos no resize: {msg}")
continue
except WebSocketDisconnect:
print("Cliente desconectado")
except Exception as e:
print(f"Erro no write_to_terminal: {e}")
finally:
try:
os.close(fd)
os.kill(pid, 9)
except:
pass
try:
await asyncio.gather(read_from_terminal(), write_to_terminal())
except Exception as e:
print(f"Erro na conexão: {e}")
finally:
try:
os.close(fd)
os.kill(pid, 9)
except:
pass |