File size: 2,608 Bytes
1b61dd2 a837371 87ba64d 16ce1dc bcbd987 5996f29 a837371 87ba64d 16ce1dc 8950a29 5996f29 a837371 87ba64d a837371 8950a29 1b61dd2 16ce1dc 87ba64d 16ce1dc 87ba64d 16ce1dc 87ba64d 16ce1dc 87ba64d 16ce1dc 87ba64d 16ce1dc 87ba64d 16ce1dc 87ba64d 16ce1dc 87ba64d 16ce1dc 87ba64d 16ce1dc 1b61dd2 87ba64d 16ce1dc 87ba64d 16ce1dc 1b61dd2 87ba64d 1b61dd2 87ba64d 16ce1dc 1b61dd2 16ce1dc 1b61dd2 87ba64d 16ce1dc 1b61dd2 87ba64d 1b61dd2 16ce1dc 87ba64d 1b61dd2 87ba64d 16ce1dc 87ba64d 16ce1dc 87ba64d 16ce1dc 1b61dd2 3adc1e7 87ba64d 3adc1e7 | 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 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | from fastapi import FastAPI, WebSocket
from fastapi.responses import PlainTextResponse
import asyncio
import os
import pty
import subprocess
app = FastAPI()
@app.get("/", response_class=PlainTextResponse)
async def root():
return "Bash WebSocket Terminal OK"
@app.get("/health", response_class=PlainTextResponse)
async def health():
return "OK"
@app.websocket("/terminals/websocket/{terminal_id}")
async def terminal(websocket: WebSocket, terminal_id: str):
await websocket.accept()
print("CONNECTED:", terminal_id, flush=True)
master, slave = pty.openpty()
process = subprocess.Popen(
["/bin/bash", "--noprofile", "--norc", "-i"],
stdin=slave,
stdout=slave,
stderr=slave,
start_new_session=True
)
os.close(slave)
loop = asyncio.get_running_loop()
async def bash_to_websocket():
while process.poll() is None:
try:
data = await loop.run_in_executor(
None,
os.read,
master,
8192
)
if not data:
break
output = data.decode(
"utf-8",
errors="replace"
)
print(
"BASH OUTPUT:",
repr(output),
flush=True
)
await websocket.send_text(output)
except Exception as e:
print(
"OUTPUT ERROR:",
repr(e),
flush=True
)
break
output_task = asyncio.create_task(
bash_to_websocket()
)
try:
while True:
data = await websocket.receive_text()
print(
"TERMUX INPUT:",
repr(data),
flush=True
)
os.write(
master,
data.encode("utf-8")
)
except Exception as e:
print(
"WEBSOCKET:",
repr(e),
flush=True
)
finally:
output_task.cancel()
try:
process.terminate()
except:
pass
try:
os.close(master)
except:
pass
print(
"DISCONNECTED:",
terminal_id,
flush=True
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(
app,
host="0.0.0.0",
port=7860
) |