import asyncio
import fcntl
import json
import os
import pty
import signal
import struct
import termios
import threading
from websockets.asyncio.server import serve
def set_winsize(fd, rows, cols):
"""Set terminal window size via TIOCSWINSZ."""
winsize = struct.pack("HHHH", rows, cols, 0, 0)
fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize)
async def shell(ws):
# Create PTY; child process gets it as controlling terminal
pid, fd = pty.fork()
if pid == 0:
os.environ["TERM"] = "xterm-256color"
os.execvp("bash", ["bash"])
# Set initial terminal size (client overrides via resize message)
rows, cols = 24, 80
set_winsize(fd, rows, cols)
q = asyncio.Queue()
def reader():
while True:
try:
data = os.read(fd, 65536)
if not data:
break
q.put_nowait(data)
except Exception:
break
rt = threading.Thread(target=reader, daemon=True)
rt.start()
try:
async def sender():
while True:
data = await q.get()
await ws.send(data)
async def receiver():
nonlocal rows, cols
async for raw in ws:
if isinstance(raw, str):
# Text frames: try JSON resize command first
try:
msg = json.loads(raw)
if "rows" in msg and "cols" in msg:
rows, cols = msg["rows"], msg["cols"]
set_winsize(fd, rows, cols)
os.kill(pid, signal.SIGWINCH)
continue
except (json.JSONDecodeError, ValueError):
pass
# Plain text from wscat: no trailing newline, append one
os.write(fd, raw.encode() + b"\n")
else:
# Binary frames: raw keystrokes from TUI client, pass through
os.write(fd, raw)
await asyncio.gather(sender(), receiver())
finally:
os.close(fd)
os.waitpid(pid, 0)
HTML_PAGE = """
WebSocket Shell
WebSocket Shell
Connect with wspty:
./wspty wss://HOST
Or with wscat (basic, no TUIs):
wscat -c wss://HOST
Source
"""
async def process_request(connection, request):
if request.headers.get("Upgrade", "").lower() != "websocket":
if request.path == "/healthz":
return connection.respond(200, "OK")
return connection.respond(200, HTML_PAGE)
async def main():
async with serve(shell, "0.0.0.0", 7860,
ping_interval=None, max_size=2**20,
process_request=process_request) as srv:
await srv.serve_forever()
asyncio.run(main())