简介这是一份面向Python Web开发者与运维人员的实战项目源码借助Django框架在浏览器中复刻Xshell的远程终端能力让用户无需安装桌面客户端即可通过Web界面与远程服务器进行SSH交互。项目围绕Django的模型、视图、模板与URL配置四大核心组件展开并借助paramiko库实现SSHv2协议的加密连接、命令执行与结果回传同时涉及用户认证、权限控制与操作日志等安全设计适合希望深入理解Web终端与网络安全的进阶学习者。资源包共29个文件以20个py源码为主辅以js、html、css等前端资源及md说明文档压缩后约85KB目录结构清晰涵盖项目配置、应用模块与静态资源。目前已有1279人学习下载读者可从中获得完整的Django项目骨架、SSH连接实现思路与终端交互界面参考便于二次开发与技能提升。1. 浏览器里跑一个“xshell”这套 Django 方案到底能解决什么很多做运维平台或内部工具的同学都遇到过同一个尴尬手里一堆 Linux 机器日常巡检、查日志、重启服务每次都要开本地终端软件切来切去账号密码散落在各个工具里想审计谁在什么时候执行了什么命令基本靠人肉回忆。更麻烦的是团队里非运维岗的同事临时要看一眼日志你还得给他配密钥、教他用终端沟通成本比干活还高。这套「Python-通过Django在web上实现xshell的功能」的资源核心就是解决这件事用 Django 做后端把 SSH 连接、命令执行、结果回显搬到浏览器里做一个轻量级的 Web 终端。它适合三类人——想给内部运维平台加一个网页终端模块的后端开发、需要统一入口管理多台机器的运维、以及正在学 Django 通道Channels和 WebSocket 实战的进阶学习者。说白了它不是一个要替代专业终端的产品而是一个能嵌进你自己系统里的“连接层”。2. 拆开看架构Django Channels Paramiko 是怎么把终端搬进浏览器的2.1 为什么是 Channels 而不是普通视图普通 Django 视图是“请求-响应”模型客户端发一个 HTTP 请求服务端处理完返回连接就断了。但终端交互是长连接、双向、持续输出的场景你敲一个字符要发到服务端服务端执行命令后要实时把输出推回来中间可能持续几十秒。用 HTTP 轮询去做延迟高、连接开销大体验会很差。Django Channels 在 Django 之上加了一层 ASGI 处理支持 WebSocket 协议正好匹配这种双向流式通信。浏览器端用 WebSocket 连上来Channels 的 Consumer 负责维持这个连接Paramiko 在服务端建立到目标 Linux 机器的 SSH 会话两边数据通过 Consumer 中转。这个分工要理清楚Channels 管“浏览器到 Django”这一段Paramiko 管“Django 到目标机”这一段Consumer 是中间的桥。常见做法是把 Consumer 写成异步的用async_to_sync包装 Paramiko 的阻塞调用或者干脆把 SSH 操作丢到线程池里执行避免阻塞事件循环。这一点如果处理不好会出现“一个用户连上来其他用户全部卡住”的情况后面避坑章节会细说。2.2 环境准备与依赖安装先把项目骨架搭起来。假设你已经有一个 Django 项目如果没有用下面的命令创建。注意 Channels 对 Django 版本有要求Django 4.x 配 Channels 4.x 是比较稳的组合。# 创建虚拟环境避免污染全局包 python -m venv venv source venv/bin/activate # Windows 下用 venv\Scripts\activate # 安装核心依赖 pip install django4.2 pip install channels4.0 pip install channels-redis4.1 pip install paramiko3.4 # 创建项目和应用 django-admin startproject webterminal cd webterminal python manage.py startapp terminal参数说明channels-redis不是必须的但如果你打算多进程部署比如用 Daphne 起多个 worker就必须用 Redis 做通道层channel layer否则 WebSocket 消息在不同进程间无法路由。开发阶段可以用InMemoryChannelLayer但上线前一定要换掉。装完之后在settings.py里注册 Channels 并配置 ASGI 应用# settings.py INSTALLED_APPS [ daphne, # 必须放在 channels 前面用于接管 runserver django.contrib.admin, django.contrib.auth, django.contrib.contenttypes, django.contrib.sessions, django.contrib.messages, django.contrib.staticfiles, channels, terminal, ] ASGI_APPLICATION webterminal.asgi.application # 开发阶段用内存通道层上线换 Redis CHANNEL_LAYERS { default: { BACKEND: channels.layers.InMemoryChannelLayer, }, }逻辑说明把daphne放在INSTALLED_APPS首位是为了让runserver命令被 Daphne 接管这样开发时也能直接处理 WebSocket 请求不用额外起 ASGI 服务器。ASGI_APPLICATION指向项目的 asgi 配置Channels 会从这里读取路由。2.3 路由与 Consumer 的编写先配置 ASGI 路由把 WebSocket 请求指向我们写的 Consumer# webterminal/asgi.py import os from django.core.asgi import get_asgi_application from channels.routing import ProtocolTypeRouter, URLRouter from channels.auth import AuthMiddlewareStack from terminal import routing os.environ.setdefault(DJANGO_SETTINGS_MODULE, webterminal.settings) application ProtocolTypeRouter({ http: get_asgi_application(), websocket: AuthMiddlewareStack( URLRouter(routing.websocket_urlpatterns) ), })# terminal/routing.py from django.urls import re_path from . import consumers websocket_urlpatterns [ re_path(rws/terminal/(?Phost_id\d)/$, consumers.TerminalConsumer.as_asgi()), ]参数说明URL 里带了host_id表示每台被管理的机器对应一个 WebSocket 地址。AuthMiddlewareStack会把 Django 的 session 认证信息带进 Consumer这样你可以在connect方法里判断用户有没有登录、有没有权限操作这台机器。生产环境一定要加权限校验否则任何人猜到 host_id 就能连上。接下来是核心的 Consumer# terminal/consumers.py import json import paramiko from channels.generic.websocket import AsyncWebsocketConsumer from asgiref.sync import sync_to_async class TerminalConsumer(AsyncWebsocketConsumer): async def connect(self): self.host_id self.scope[url_route][kwargs][host_id] # 校验登录状态未登录直接拒绝 if not self.scope[user].is_authenticated: await self.close() return await self.accept() self.ssh_client None self.channel None async def receive(self, text_data): data json.loads(text_data) action data.get(action) if action connect: # 建立 SSH 连接放到线程池执行避免阻塞 await self._open_ssh(data) elif action input: # 把用户输入写到 SSH 通道 if self.channel: await sync_to_async(self.channel.send)(data[data]) elif action resize: # 调整伪终端窗口大小 if self.channel: await sync_to_async(self.channel.resize_pty)( widthdata[cols], heightdata[rows] ) async def _open_ssh(self, data): def _connect(): client paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) client.connect( hostnamedata[host], portdata.get(port, 22), usernamedata[username], passworddata.get(password), key_filenamedata.get(key_path), timeout10, ) chan client.invoke_shell(termxterm) chan.settimeout(0.0) return client, chan try: self.ssh_client, self.channel await sync_to_async(_connect)() await self.send(json.dumps({type: connected})) # 启动循环读取 SSH 输出 import asyncio asyncio.create_task(self._read_loop()) except Exception as e: await self.send(json.dumps({type: error, message: str(e)})) async def _read_loop(self): import asyncio while True: if self.channel is None: break try: # recv 是阻塞的用 sync_to_async 包装 data await sync_to_async(self.channel.recv)(4096) if not data: break await self.send(json.dumps({ type: output, data: data.decode(utf-8, errorsreplace) })) except Exception: break await asyncio.sleep(0.01) async def disconnect(self, close_code): if self.channel: self.channel.close() if self.ssh_client: self.ssh_client.close()逻辑说明connect里做认证receive里根据前端发来的 action 分发处理。_open_ssh把 Paramiko 的连接和invoke_shell放在sync_to_async里执行因为 Paramiko 是同步阻塞库直接在异步函数里调用会卡住整个事件循环。_read_loop是一个持续读取 SSH 输出的任务读到数据就通过 WebSocket 推给前端。disconnect里要记得关闭通道和客户端否则连接会泄漏。参数说明invoke_shell(termxterm)里的 term 类型影响终端转义序列的解析前端 xterm.js 一般用 xterm 或 xterm-256color。settimeout(0.0)把通道设为非阻塞模式配合循环读取。recv(4096)每次读 4KB太小会增加循环次数太大可能延迟回显4096 是个折中值。2.4 前端 xterm.js 的接入前端用 xterm.js 渲染终端界面它负责把用户按键转成数据发给 WebSocket同时把服务端推来的输出写进屏幕。// static/js/terminal.js import { Terminal } from xterm; import { FitAddon } from xterm-addon-fit; import xterm/css/xterm.css; const term new Terminal({ cursorBlink: true, fontSize: 14, fontFamily: Menlo, Monaco, monospace, theme: { background: #1e1e1e, foreground: #d4d4d4 } }); const fitAddon new FitAddon(); term.loadAddon(fitAddon); term.open(document.getElementById(terminal)); fitAddon.fit(); const hostId document.getElementById(terminal).dataset.hostId; const ws new WebSocket(ws://${window.location.host}/ws/terminal/${hostId}/); ws.onopen () { // 连接建立后发送 SSH 认证信息 ws.send(JSON.stringify({ action: connect, host: 192.168.1.10, port: 22, username: ops, password: your-password })); }; ws.onmessage (event) { const msg JSON.parse(event.data); if (msg.type output) { term.write(msg.data); } else if (msg.type error) { term.write(\r\n\x1b[31m连接失败: ${msg.message}\x1b[0m\r\n); } }; // 用户按键 - 发给服务端 term.onData((data) { if (ws.readyState WebSocket.OPEN) { ws.send(JSON.stringify({ action: input, data: data })); } }); // 窗口大小变化时同步给服务端 window.addEventListener(resize, () { fitAddon.fit(); if (ws.readyState WebSocket.OPEN) { ws.send(JSON.stringify({ action: resize, cols: term.cols, rows: term.rows })); } });逻辑说明term.onData捕获用户的每一次按键包括回车、退格、CtrlC 等控制字符原样发给服务端。服务端把这些字符写进 SSH 通道远端 shell 解析后返回输出再通过term.write渲染。resize事件同步终端尺寸否则vim、top这类全屏程序会显示错乱。参数说明fitAddon.fit()根据容器大小自动计算行列数容器必须有明确的宽高否则算出来是 0。密码在前端明文传输是个隐患生产环境应该改成后端根据 host_id 从数据库取凭据前端只传 host_id不传密码。3. 把连接管起来主机配置、凭据管理与多会话隔离3.1 主机信息建模把机器信息存到数据库前端只传 host_id凭据由后端查表获取这样既安全又方便管理。# terminal/models.py from django.db import models from django.contrib.auth.models import User class Host(models.Model): name models.CharField(max_length64, verbose_name主机别名) ip models.GenericIPAddressField(verbose_nameIP地址) port models.IntegerField(default22, verbose_nameSSH端口) username models.CharField(max_length64, verbose_name登录用户) auth_type models.CharField( max_length10, choices[(password, 密码), (key, 密钥)], defaultpassword ) password models.CharField(max_length256, blankTrue, verbose_name密码) key_file models.FileField(upload_tokeys/, blankTrue, verbose_name密钥文件) owner models.ForeignKey(User, on_deletemodels.CASCADE, verbose_name归属用户) created_at models.DateTimeField(auto_now_addTrue) class Meta: verbose_name 主机 verbose_name_plural 主机参数说明auth_type区分密码和密钥两种认证方式key_file用 FileField 存上传的私钥。密码字段实际存储时应该加密Django 没有内置的字段级加密常见做法是用cryptography库的 Fernet 做对称加密或者接入 KMS。owner字段用于权限隔离查询时过滤ownerrequest.user防止越权访问别人的机器。3.2 凭据加密与取用# terminal/crypto.py import base64 import os from cryptography.fernet import Fernet from django.conf import settings def _get_fernet(): # 密钥从环境变量读取不要硬编码在代码里 key settings.SSH_CRED_KEY.encode() return Fernet(key) def encrypt_password(plain: str) - str: f _get_fernet() return f.encrypt(plain.encode()).decode() def decrypt_password(token: str) - str: f _get_fernet() return f.decrypt(token.encode()).decode()逻辑说明Fernet 是对称加密密钥必须固定且保密否则重启后解不开之前存的数据。密钥生成用Fernet.generate_key()生成后放到环境变量或配置中心不要提交到代码仓库。取用凭据时在 Consumer 里根据 host_id 查库、解密再传给 Paramiko。3.3 多会话隔离与资源限制每个 WebSocket 连接对应一个独立的 Consumer 实例实例之间天然隔离。但要注意两点一是同一个用户可能开多个标签页连同一台机器每个连接都会新建一个 SSH 会话目标机的sshd有 MaxSessions 限制开太多会被拒绝二是如果某个会话执行了tail -f这类持续输出的命令读取循环会一直跑占用线程池资源。常见做法是加一个会话数限制在connect时检查当前用户已建立的连接数超过阈值就拒绝。另外给_read_loop加一个空闲超时比如 30 分钟没有输入就主动断开避免僵尸会话堆积。# 在 Consumer 里加空闲检测 import time async def _read_loop(self): last_active time.time() while True: if time.time() - last_active 1800: await self.send(json.dumps({type: error, message: 会话超时})) await self.close() break # ... 原有读取逻辑 if data: last_active time.time()参数说明1800 秒是 30 分钟按团队实际使用频率调整。如果用户经常挂着终端看监控可以调大或者改成只在无输出时计时。4. 避坑与排查那些让我加班到凌晨的细节4.1 现象一个用户连上后其他用户全部卡死原因Paramiko 的recv是阻塞调用如果直接在异步的_read_loop里调用而没有用sync_to_async包装整个事件循环会被卡住所有 WebSocket 连接都无法处理消息。解决所有 Paramiko 的阻塞方法connect、recv、send、resize_pty都必须用sync_to_async包装。另外sync_to_async默认在线程池里执行线程池大小有限如果并发连接很多需要调整ASGI_THREADS环境变量或者改用ThreadPoolExecutor自己管理。4.2 现象终端里vim打开后花屏方向键变成^[[A原因前端 xterm.js 的 term 类型和服务端invoke_shell的 term 不匹配或者前端没有正确处理转义序列。另外resize没有同步导致远端以为终端还是默认的 80x24。解决确保invoke_shell(termxterm-256color)前端 xterm.js 初始化时也设置termName: xterm-256color。在connect成功后立即发送一次resize把当前的行列数同步过去。如果还有花屏检查 WebSocket 传输时有没有对特殊字符做转义json.dumps默认会转义非 ASCII但控制字符是 ASCII 范围内的一般没问题。4.3 现象连接几秒后自动断开日志显示Connection reset by peer原因目标机的sshd配置了ClientAliveInterval如果一段时间没有数据交互会主动断开。另外有些云主机的安全组对空闲连接有超时限制。解决在 Paramiko 的Transport上开启 keepaliveclient paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) client.connect(...) transport client.get_transport() transport.set_keepalive(30) # 每 30 秒发一次心跳参数说明30 秒是常见值要小于目标机ClientAliveInterval的设置。如果目标机是 60 秒设 30 秒就够。同时在前端加一个心跳每 20 秒发一个空消息保持 WebSocket 活跃。4.4 现象上传密钥文件后连接报not a valid RSA private key file原因Paramiko 对密钥格式有要求OpenSSH 新格式以-----BEGIN OPENSSH PRIVATE KEY-----开头在旧版本 Paramiko 里不支持需要转成 PEM 格式。另外密钥文件权限不对也会报错。解决用ssh-keygen -p -m PEM -f your_key把密钥转成 PEM 格式。或者在代码里用paramiko.RSAKey.from_private_key_file()显式指定类型。文件权限在服务端保存时设为 600Paramiko 会检查。4.5 现象生产环境用 InMemoryChannelLayer多 worker 下 WebSocket 消息丢失原因InMemoryChannelLayer只在单个进程内有效Daphne 起多个 worker 时WebSocket 连接可能落在 worker A而消息从 worker B 发出B 找不到 A 的通道消息就丢了。解决换成channels_redisCHANNEL_LAYERS { default: { BACKEND: channels_redis.core.RedisChannelLayer, CONFIG: { hosts: [(127.0.0.1, 6379)], }, }, }参数说明Redis 地址按实际部署填如果有密码加在 hosts 元组里。上线前务必确认 Redis 连通性否则 WebSocket 会直接连不上。5. 进阶把命令审计和会话回放做进去基础功能跑通之后真正让这套东西在团队里站住脚的是审计能力。运维平台没有审计等于没有后悔药。我一般会在 Consumer 的receive里加一层记录用户每次发送input时把数据和时间戳写进数据库同时把服务端返回的output也存一份按会话 ID 关联。这样事后可以按人、按机器、按时间段查“谁在什么时候执行了什么”。# terminal/models.py 追加 class CommandLog(models.Model): session_id models.CharField(max_length64, db_indexTrue) user models.ForeignKey(User, on_deletemodels.CASCADE) host models.ForeignKey(Host, on_deletemodels.CASCADE) direction models.CharField(max_length6) # input / output content models.TextField() created_at models.DateTimeField(auto_now_addTrue, db_indexTrue) class Meta: indexes [ models.Index(fields[user, created_at]), models.Index(fields[host, created_at]), ]写入时用批量插入降低开销比如每 20 条或每 2 秒 flush 一次不要每条都写库否则高频输出会把数据库打满。会话回放就是按session_id和created_at排序把 input 和 output 按时间轴重放前端可以用一个简单的播放器按时间间隔逐条渲染。验证审计是否生效可以写一个管理命令模拟一次连接、执行几条命令、断开然后查库确认记录完整python manage.py shell -c from terminal.models import CommandLog logs CommandLog.objects.filter(session_idtest-session).order_by(created_at) for log in logs: print(log.direction, log.content[:50]) 参数说明session_id在 Consumer 的connect里生成用uuid4().hex即可贯穿整个会话。查询时加db_indexTrue的字段做过滤避免全表扫描。还有一个容易忽略的点命令审计要记录原始输入但用户可能输入密码比如mysql -p后回车输入密码这些敏感内容不应该明文存。常见做法是对 input 做正则匹配遇到password、passwd、-p等关键词后的内容做脱敏或者干脆只记录命令本身不记录交互式输入。这个取舍看团队的安全要求但至少要有意识。从那以后我每次做 Web 终端类的项目都会先把审计表建好、把脱敏规则定好再动手写连接逻辑。因为连接层可以慢慢调审计漏了就是真的漏了补不回来。希望帮到你。本文还有配套的精品资源点击获取