| 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): |
| |
| pid, fd = pty.fork() |
| if pid == 0: |
| os.environ["TERM"] = "xterm-256color" |
| os.execvp("bash", ["bash"]) |
|
|
| |
| 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): |
| |
| 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 |
| |
| os.write(fd, raw.encode() + b"\n") |
| else: |
| |
| os.write(fd, raw) |
|
|
| await asyncio.gather(sender(), receiver()) |
| finally: |
| os.close(fd) |
| os.waitpid(pid, 0) |
|
|
|
|
| HTML_PAGE = """<!DOCTYPE html> |
| <html><head><title>WebSocket Shell</title><meta charset="utf-8"> |
| <meta name="viewport" content="width=device-width,initial-scale=1"> |
| <style>body{font:14px monospace;max-width:720px;margin:40px auto;padding:0 20px; |
| background:#0d1117;color:#c9d1d9}a{color:#58a6ff}pre{background:#161b22; |
| padding:16px;border-radius:6px;overflow-x:auto}</style></head> |
| <body><h1>WebSocket Shell</h1><p>Connect with <code>wspty</code>:</p> |
| <pre>./wspty wss://HOST</pre> |
| <p>Or with <code>wscat</code> (basic, no TUIs):</p> |
| <pre>wscat -c wss://HOST</pre> |
| <p><a href="https://github.com">Source</a></p></body></html>""" |
|
|
|
|
| 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()) |
|
|
