基于xterm.js的浏览器终端如何高效获取用户完整命令？

问题背景 #

我用xterm.js在浏览器中运行终端，将输入发送至后端Django服务，views.py代码基于MahmoudAlyy的GitHub仓库修改。当前pty_input方法会接收用户按下的所有按键，若修改该方法仅获取完整命令耗时久且易出bug，请问是否可通过修改前端或后端代码实现仅获取用户完整命令？

可行方案 #

方案一：前端缓存输入，仅在回车时发送完整命令 #

在前端xterm.js逻辑中缓存用户输入，处理退格、删除等编辑操作，直到用户按下回车键时，再将完整命令发送给后端，同时保留实时回显体验。

修改后的前端脚本（index.html）：

var socket = io.connect({ transports: ["websocket", "polling"] });

const status = document.getElementById("status")
const button = document.getElementById("button")
const fit = new FitAddon.FitAddon();

var term = new Terminal({
    cursorBlink: true,
});

term.loadAddon(fit);
term.open(document.getElementById('terminal'));
fit.fit();

// 缓存当前输入的命令
let currentCommand = '';

term.onData(data => {
    // 处理回车键：发送完整命令并重置缓存
    if (data === '\r' || data === '\n') {
        term.write('\n');
        socket.emit("pty_input", { "input": currentCommand + '\n' });
        currentCommand = '';
    } 
    // 处理退格键：删除缓存的最后一个字符并模拟终端退格效果
    else if (data === '\x7f') {
        if (currentCommand.length > 0) {
            currentCommand = currentCommand.slice(0, -1);
            term.write('\x1b[D \x1b[D');
        }
    } 
    // 处理普通字符：添加到缓存并回显
    else {
        currentCommand += data;
        term.write(data);
    }
});

socket.on("pty_output", function (output) {
    term.write(output["output"])
})

socket.on("connect", () => {
    status.innerHTML = '<span style="background-color: lightgreen;">connected</span>'
    button.innerHTML = 'Disconnect'
})

socket.on("disconnect", () => {
    status.innerHTML = '<span style="background-color: #ff8383;">disconnected</span>'
    button.innerHTML = 'Connect'

})

function myFunction() {
    if (button.innerHTML == 'Connect') {
        location.reload();
    }

    else if (button.innerHTML == "Disconnect") {
        socket.emit("disconnect_request")
    }
}

function resize() {
    term.fit()
    socket.emit("resize", { "cols": term.cols, "rows": term.rows })
}

window.onresize = resize
window.onload = resize

方案二：后端通过PTY输出解析完整命令 #

不修改pty_input逻辑，在读取PTY输出时，通过识别shell提示符和换行符来提取用户输入的完整命令，适合后端统一处理命令的场景。

修改后的后端代码（views.py）：

import struct
import fcntl
import termios
import select
import os
import subprocess
import signal
import socketio

async_mode = "eventlet"
sio = socketio.Server(async_mode=async_mode)

fd = None
child_pid = None
# 跟踪命令行状态
current_command = ''
in_prompt = True  # 是否处于等待命令的提示符状态
# 常见shell提示符结尾，可根据实际场景调整
prompt_endings = ['$', '#', '%']

def index(request):
    return render(request, "index.html")

def set_winsize(fd, row, col, xpix=0, ypix=0):
    winsize = struct.pack("HHHH", row, col, xpix, ypix)
    fcntl.ioctl(fd, termios.TIOCSWINSZ, winsize)

def read_and_forward_pty_output():
    global fd, current_command, in_prompt
    max_read_bytes = 1024 * 20

    while True:
        sio.sleep(0.01)
        if fd:
            timeout_sec = 0
            (data_ready, _, _) = select.select([fd], [], [], timeout_sec)
            if data_ready:
                output = os.read(fd, max_read_bytes).decode()
                sio.emit("pty_output", {"output": output})

                # 解析输出提取完整命令
                for char in output:
                    if in_prompt:
                        # 检测到提示符结尾，开始记录命令
                        if char in prompt_endings:
                            in_prompt = False
                    else:
                        if char == '\n':
                            # 命令结束，可在此处处理完整命令（如审计、校验）
                            print(f"捕获完整命令: {current_command.strip()}")
                            current_command = ''
                            in_prompt = True
                        else:
                            current_command += char
        else:
            print("process killed")
            return

# 其余原有函数保持不变
@sio.event
def resize(sid, message):
    if fd:
        set_winsize(fd, message["rows"], message["cols"])

@sio.event
def pty_input(sid, message):
    if fd:
        os.write(fd, message["input"].encode())

@sio.event
def disconnect_request(sid):
    sio.disconnect(sid)

@sio.event
def connect(sid, environ):
    global fd
    global child_pid

    if child_pid:
        os.write(fd, "\n".encode())
        return

    (child_pid, fd) = pty.fork()

    if child_pid == 0:
        subprocess.run("bash")

    else:
        print()
        sio.start_background_task(target=read_and_forward_pty_output)

@sio.event
def disconnect(sid):
    global fd
    global child_pid

    os.kill(child_pid, signal.SIGKILL)
    os.wait()

    fd = None
    child_pid = None
    print("Client disconnected")

方案对比 #

• 前端方案：逻辑简单，无需后端修改，适合仅需前端控制命令发送时机的场景，但需处理各类终端编辑操作（如光标移动、删除），否则会出现缓存与实际输入不一致的问题。
• 后端方案：无需修改前端，能在后端统一捕获命令，适合命令审计、权限校验等场景，但依赖shell提示符格式，若用户自定义提示符需调整检测逻辑。

内容的提问来源于stack exchange，提问作者ivenoidea