bridge.py

import asyncio
import os
import pty
import select


class TerminalBridge:

    def __init__(self):
        self.sessions = {}

    async def connect(self, websocket):

        pid, fd = pty.fork()

        if pid == 0:
            import os

if pid == 0:
    os.makedirs("/data/workspace", exist_ok=True)
    os.chdir("/data/workspace")

    os.execvp(
        "bash",
        [
            "bash",
            "--login",
            "-i"
        ]
    )

        session = {
            "ws": websocket,
            "pid": pid,
            "fd": fd
        }

        self.sessions[websocket] = session

        asyncio.create_task(self.reader(session))

        return session

    async def disconnect(self, session):

        try:
            os.close(session["fd"])
        except Exception:
            pass

        self.sessions.pop(session["ws"], None)

    async def input(self, session, text):

        os.write(
            session["fd"],
            (text + "\n").encode()
        )

    async def reader(self, session):

        fd = session["fd"]

        while True:

            ready, _, _ = select.select(
                [fd],
                [],
                [],
                0.05
            )

            if ready:

                try:

                    data = os.read(
                        fd,
                        4096
                    )

                    if not data:
                        break

                    await session["ws"].send_text(
                        data.decode(errors="ignore")
                    )

                except Exception:
                    break