本文详解如何在 Django 后端中为每个浏览器客户端动态创建独立的伪终端（PTY）会话，避免多个用户共享同一终端实例的问题，并提供基于 PtyTerminal 封装类与 Socket.IO 协同工作的生产级实现方案。
本文详解如何在 django 后端中为每个浏览器客户端动态创建独立的伪终端（pty）会话，避免多个用户共享同一终端实例的问题，并提供基于 `ptyterminal` 封装类与 socket.io 协同工作的生产级实现方案。
当前代码的核心问题是全局共享 fd 和 child_pid 变量，导致所有客户端连接都复用同一个 PTY 进程——这本质上是单例模式误用于并发场景。解决思路必须转向会话隔离：为每个 Socket.IO 连接分配独立的 PTY 实例，并通过唯一会话标识（如 sid）进行生命周期管理。
✅ 步骤一：封装 PTY 操作为可复用类
首先将原始混乱的 pty 逻辑抽象为线程安全、资源可控的 PtyTerminal 类（已优化适配 Django 环境）：
# pty_terminal.py
import os
import signal
from select import select
from pty import fork
from termios import TIOCSWINSZ, TIOCGWINSZ
from struct import pack, unpack
from fcntl import ioctl
from os import execl, closerange
from time import sleep
DEFAULT_SIZE = (24, 80, 0, 0)
DEFAULT_CMD = ["/bin/bash"] # 替换为你需要的 shell 或命令
class PtyTerminal:
def __init__(self, cmd=None, size=None):
self.master = None
self.child = None
self._size = DEFAULT_SIZE
self.exit_code = None
child_pid, master_fd = fork()
if child_pid == 0: # 子进程
closerange(3, 256)
os.setsid()
os.dup2(master_fd, 0) # stdin
os.dup2(master_fd, 1) # stdout
os.dup2(master_fd, 2) # stderr
os.close(master_fd)
execl(*(cmd or DEFAULT_CMD))
else: # 父进程
self.child = child_pid
self.master = master_fd
self.size = size or DEFAULT_SIZE
@property
def size(self):
return self._size
@size.setter
def size(self, value):
if self.master:
ioctl(self.master, TIOCSWINSZ, pack("HHHH", *value))
try:
self._size = unpack("HHHH", ioctl(self.master, TIOCGWINSZ, b"\x00" * 8))
except OSError:
pass
def read(self, length=4096, timeout=0.01):
if not self.master:
return b""
try:
rd, _, _ = select([self.master], [], [], timeout)
if rd:
return os.read(self.master, length)
except (OSError, ValueError):
pass
return b""
def write(self, data):
if self.master and data:
if isinstance(data, str):
data = data.encode("utf-8")
try:
os.write(self.master, data)
except OSError:
pass
def close(self):
if self.master:
os.close(self.master)
self.master = None
if self.child:
try:
os.kill(self.child, signal.SIGTERM)
os.waitpid(self.child, 0)
except (OSError, ChildProcessError):
pass
self.child = None
该类确保每个终端实例拥有独立的文件描述符、子进程 PID 和窗口尺寸控制能力，且支持安全读写与资源释放。
✅ 步骤二：按会话隔离管理终端实例
修改 views.py，移除全局 fd/child_pid，改用字典 terminals: Dict[str, PtyTerminal] 映射 sid → terminal：
# views.py
import os
import signal
from django.shortcuts import render
import socketio
from eventlet import sleep
from .pty_terminal import PtyTerminal
async_mode = "eventlet"
sio = socketio.Server(async_mode=async_mode)
# 存储每个 sid 对应的终端实例
terminals = {}
def index(request):
return render(request, "index.html")
@sio.event
def connect(sid, environ):
# 为每个新连接创建独立终端
terminal = PtyTerminal()
terminals[sid] = terminal
print(f"[+] New terminal session created for {sid}")
# 启动后台任务持续推送输出
sio.start_background_task(target=forward_output, sid=sid, terminal=terminal)
@sio.event
def disconnect(sid):
if sid in terminals:
terminals[sid].close()
del terminals[sid]
print(f"[-] Terminal session closed for {sid}")
@sio.event
def pty_input(sid, data):
if sid in terminals:
terminals[sid].write(data.get("input", ""))
@sio.event
def resize(sid, message):
if sid in terminals:
rows = message.get("rows", 24)
cols = message.get("cols", 80)
terminals[sid].size = (rows, cols, 0, 0)
def forward_output(sid, terminal):
"""向客户端持续推送终端输出"""
while sid in terminals:
output = terminal.read(timeout=0.02)
if output:
try:
sio.emit("pty_output", {"output": output.decode("utf-8")}, room=sid)
except UnicodeDecodeError:
sio.emit("pty_output", {"output": output.decode("utf-8", errors="replace")}, room=sid)
sleep(0.01)
⚠️ 关键改进点：
- room=sid 参数确保 pty_output 仅广播给当前会话，杜绝跨用户污染；
- 每个 sid 绑定专属 PtyTerminal 实例，彻底解耦；
- forward_output 作为协程任务运行，避免阻塞主线程；
- disconnect 中显式调用 .close()，防止僵尸进程和 fd 泄漏。
✅ 步骤三：前端适配 —— 确保会话绑定正确
index.html 中无需修改核心逻辑，但需确认 Socket.IO 客户端未复用连接（尤其多标签场景）。建议在 connect 前清除旧连接：
// 在 myFunction() 中增强连接逻辑
function myFunction() {
if (button.innerHTML === 'Connect') {
// 断开已有连接（如有）
if (socket.connected) {
socket.disconnect();
}
socket.connect();
} else if (button.innerHTML === "Disconnect") {
socket.emit("disconnect_request");
socket.disconnect();
}
}
同时，服务端 connect 事件中无需返回 token，因为 Socket.IO 自带 sid 全局唯一性保障，天然适配会话隔离。
? 补充建议与注意事项
- 安全加固：在 connect 中加入权限校验（如登录态验证、IP 限频），防止恶意创建大量 PTY 耗尽系统资源；
- 超时清理：添加定时任务扫描 terminals 字典，对空闲超过 5 分钟的会话执行 force_kill_close()；
- 资源监控：记录每个终端的 CPU/内存占用，异常时自动终止；
- 命令限制：通过 PtyTerminal(cmd=...) 传入白名单命令（如 ["/usr/bin/python3", "-i"]），禁用危险 shell 操作；
- 编码兼容：read() 返回 bytes，务必用 decode('utf-8', errors='replace') 防止乱码中断渲染。
通过以上重构，你的 Django + xterm.js 终端应用即可完美支持任意数量并发用户，每个用户获得完全独立、互不干扰的终端体验，真正实现企业级多租户终端服务能力。
