admin.py

#!/usr/bin/env python3
"""
keycoms — admin.py
Hosts the WebSocket server. Captures every keystroke and streams to client.
Receives keystrokes from client and prints them in terminal only.

Usage:
  python3 admin.py [port]   default port: 8765
  Then share your IP/tunnel URL with the client.

Controls:
  ESC   — disconnect client and shut down
"""

import asyncio
import json
import sys
import threading

try:
    import keyboard
except ImportError:
    print("[admin] Missing: pip install keyboard")
    sys.exit(1)

try:
    from websockets.asyncio.server import serve
    from websockets.exceptions import ConnectionClosed
except ImportError:
    print("[admin] Missing: pip install websockets")
    sys.exit(1)

PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8765

# ── Shared state ───────────────────────────────────────────────────────────────
state = {
    "client_ws": None,
    "loop":      None,
    "running":   True,
    "recv_buf":  [],
}

# ── Display helpers ────────────────────────────────────────────────────────────
DISPLAY_MAP = {"\b": "⌫", "\n": "↵\n", "\t": "→"}

def display_char(value: str) -> str:
    return DISPLAY_MAP.get(value, value)


# ── Keystroke capture (admin → client) ────────────────────────────────────────
def on_key(event):
    if event.event_type != keyboard.KEY_DOWN:
        return

    name = event.name

    if name == "esc":
        print("\n[admin] ESC — shutting down...")
        state["running"] = False
        if state["client_ws"] and state["loop"]:
            asyncio.run_coroutine_threadsafe(
                state["client_ws"].close(), state["loop"]
            )
        return

    KEY_MAP = {
        "space":     " ",
        "enter":     "\n",
        "backspace": "\b",
        "tab":       "\t",
    }

    if name in KEY_MAP:
        value = KEY_MAP[name]
    elif len(name) == 1:
        value = name
    else:
        return

    ws = state["client_ws"]
    loop = state["loop"]
    if ws and loop:
        try:
            asyncio.run_coroutine_threadsafe(
                ws.send(json.dumps({"type": "key", "from": "admin", "value": value})),
                loop
            )
        except Exception:
            pass


def start_capture():
    keyboard.hook(on_key)
    keyboard.wait()


# ── WebSocket handler ──────────────────────────────────────────────────────────
async def handler(ws):
    if state["client_ws"] is not None:
        await ws.send(json.dumps({"type": "error", "msg": "Session occupied."}))
        await ws.close()
        return

    state["client_ws"] = ws
    print(f"[admin] Client connected: {ws.remote_address}")
    print("[admin] Start typing anywhere — keystrokes stream to client.")
    print("[admin] Client keystrokes appear below. ESC to disconnect.\n")
    print("[admin] ── incoming from client ──────────────────────────")

    try:
        async for raw in ws:
            try:
                data = json.loads(raw)
            except json.JSONDecodeError:
                continue

            if data.get("type") == "key":
                value = data.get("value", "")
                print(display_char(value), end="", flush=True)

    except ConnectionClosed:
        pass
    finally:
        state["client_ws"] = None
        print("\n[admin] Client disconnected.")


# ── Main ───────────────────────────────────────────────────────────────────────
async def main():
    print("╔═══════════════════════════════╗")
    print("║        keycoms  admin         ║")
    print("╚═══════════════════════════════╝")
    print(f"[admin] Listening on ws://0.0.0.0:{PORT}")
    print(f"[admin] Tunnel: cloudflared tunnel --url ws://localhost:{PORT}\n")

    state["loop"] = asyncio.get_running_loop()

    t = threading.Thread(target=start_capture, daemon=True)
    t.start()

    async with serve(handler, "0.0.0.0", PORT):
        while state["running"]:
            await asyncio.sleep(0.1)

    print("[admin] Server stopped.")


if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n[admin] Interrupted.")