简介针对有Python基础、需落地主机远程访问安全管控的开发者或运维人员这份资料围绕Python编写堡垒机给出系统性知识梳理。堡垒机即安全运维审计系统资源覆盖SSH加密通信socket与paramiko、身份认证与公钥机制、基于角色的权限管理、操作会话审计、多线程与异步并发、Web管理界面、数据持久化、安全策略及API集成等模块。具体涉及paramiko处理加密登录、getpass规避口令明文暴露、logging记录命令与结果、RBAC控制访问边界、threading/asyncio提升并发响应、Flask/Django构建操作界面、SQLite/PostgreSQL存储用户与日志以及失败登录限制、会话超时和pytest测试建议。内容以知识点拆分为主适合从零搭建轻量级堡垒机或为已有工具补强审计能力。资源包为RAR压缩格式体积约58KB目前已有273人学习/浏览对快速理解Python安全运维工具设计有直接帮助。1. Python 写堡垒机不只是“跳板”转发、录屏与审计一个都不能少用 Python 写堡垒机很多人第一反应是做个“跳板”让用户先连到一台中转机再跳到生产服务器。真把链路搭起来才会发现能连上只是最简单的一步。堡垒机真正要解决的是三件事把访问入口收敛到一个可控节点把用户操作原样留下来以及让事后能说清楚“谁在什么时间对哪台机器做了什么”。这套东西如果交给现成商业产品通常要掏钱、要部署依赖如果用 Python 自己写paramiko 和 asyncssh 两个库就能覆盖从端口转发到会话代理的大部分场景。适合谁看有 SSH 基础、想给团队做轻量运维入口、或者单纯想弄懂堡垒机底层原理的从业者。本文从转发链路讲起给最小可复现代码再逐个拆踩坑现场。2. 堡垒机的链路模型SSH 转发、会话代理与审计先把这三件事拆开2.1 一次登录背后的三段链路客户端、Python 服务、目标机没有堡垒机时用户直接ssh user10.0.1.20客户端与目标机之间是一条点对点 SSH 连接。引入堡垒机后链路变成了三段客户端先连堡垒机堡垒机再连目标机两条独立 SSH 会话在堡垒机这里“接上头”。这里的角色划分很关键。客户端与堡垒机之间的会话本质上是用户认证和权限校验的入口堡垒机与目标机之间的会话则是实际执行命令的通道。两段会话如何对接直接决定了你能拿到什么级别的审计数据。如果只是把两段 TCP 流在网络层透传用户一次性完成 SSH 握手堡垒机只能看到加密后的字节流如果让 Python 服务同时扮演 SSH 服务端和 SSH 客户端用户先登录到你的服务你的服务再登录目标机那么用户敲的每条命令、目标机返回的每个字节都会在堡垒机内存里以明文形式经过。所以“python 写堡垒机”的链路模型本质上是在选你愿意在协议栈里站多深。站得浅代码短但审计能力弱站得深代码多但录屏、拦截、审批都能做。多数自建方案最后都落在后者因为堡垒机的核心价值恰恰在审计。2.2 选 paramiko 还是 asyncssh同步阻塞与异步并发的分界线写 Python 堡垒机绕不开两个库paramiko 和 asyncssh。paramiko 是同步阻塞风格API 贴近原生 socket调试直观网上资料多适合先跑通模型asyncssh 是 asyncio 原生实现并发能力更强但事件循环和回调式的写法会让新手绕路。我的选择标准很朴素先看要撑多少并发。内部团队几十人、上百人同时在线paramiko 加线程池完全扛得住如果要做成上千会话的网关服务再考虑 asyncssh 重写。不要一开始就上异步堡垒机最难的不是并发是会话生命周期管理和录屏完整性这些在同步模型里更容易排查。paramiko 的另一个优势是它同时实现 SSH 客户端和服务端。一个Transport对象可以 accept 客户端的连接也可以作为客户端去 connect 目标机两段逻辑用同一套 channel 抽象桥接代码可以写得很短。asyncssh 也能做同样的事但服务端和客户端的 handler 分离代码结构差异更大。2.3 为什么“TCP 转发”不是堡垒机密文透传与明文代理的区别很多人写第一版跳板用direct-tcpipchannel把客户端字节流原样转发到目标 22 端口。链路是通的但这只能叫 TCP 转发器不叫堡垒机。原因在于客户端与目标机之间完整的 SSH 加密会话穿过了转发器你的 Python 服务只是一个“网线”。它能看到握手阶段的明文 banner也能统计连接时长但用户执行了什么命令、改了什么文件完全看不到。审计只能做到连接级内容级是黑匣子。要做到内容审计必须让两段 SSH 会话在 Python 服务处“终结”再“重建”客户端登录到你的服务你的服务再作为新的 SSH 客户端登录目标机。这样密钥协商、用户认证发生在两端中间的数据在内存里是明文可以录屏、可以关键字拦截、可以审批。这就是“会话代理”与“TCP 转发”的本质区别。第 3 章先给最小转发实现让你看到链路通起来长什么样第 4 章再切换成真正能审计的会话代理。3. 最小可复现的跳板用 paramiko 把本地 2222 端口交给目标机 22 端口3.1 环境准备只装 paramiko先打通到堡垒机的 Transport先准备环境。Python 3.10 以上即可依赖只有一个 paramiko。pip install paramiko python -c import paramiko; print(paramiko.__version__)能正常打印版本号就说明环境没问题。这里不指定版本装当前稳定版即可。paramiko 依赖 cryptography 和 bcrypt安装时会自动带上不用额外处理。接下来要理解一个概念Transport是一条已建立的 SSH 连接。所有 channel 都从 transport 上打开。我们先把 Python 脚本连到堡垒机拿到 transport后续所有转发逻辑都建立在它之上。3.2 最小端口转发代码把本地 2222 交给目标机 22这段代码是一个完整的本地端口转发器。监听本地 2222 端口每进来一个 TCP 连接就通过堡垒机打开一个direct-tcpipchannel 到目标机的 22 端口然后把 socket 和 channel 双向泵起来。import socket import threading import paramiko BASTION_HOST 10.0.0.5 BASTION_PORT 22 BASTION_USER proxy BASTION_KEY /home/proxy/.ssh/id_rsa TARGET_HOST 10.0.1.20 TARGET_PORT 22 LOCAL_HOST 127.0.0.1 LOCAL_PORT 2222 def connect_bastion(): cli paramiko.SSHClient() cli.set_missing_host_key_policy(paramiko.AutoAddPolicy()) cli.connect( BASTION_HOST, portBASTION_PORT, usernameBASTION_USER, key_filenameBASTION_KEY, ) return cli.get_transport() def pump(src, dst): try: while True: data src.recv(4096) if not data: break dst.sendall(data) except Exception: pass finally: try: dst.shutdown(socket.SHUT_WR) except Exception: pass def handle_client(conn, transport): chan transport.open_channel( direct-tcpip, (TARGET_HOST, TARGET_PORT), conn.getpeername(), ) threading.Thread(targetpump, args(conn, chan), daemonTrue).start() threading.Thread(targetpump, args(chan, conn), daemonTrue).start() def main(): transport connect_bastion() listener socket.socket(socket.AF_INET, socket.SOCK_STREAM) listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) listener.bind((LOCAL_HOST, LOCAL_PORT)) listener.listen(100) print(flistening on {LOCAL_HOST}:{LOCAL_PORT}) while True: conn, _ listener.accept() threading.Thread(targethandle_client, args(conn, transport), daemonTrue).start() if __name__ __main__: main()connect_bastion()里用SSHClient.connect完成用户认证和密钥交换get_transport()把底层连接暴露出来。open_channel(direct-tcpip, ...)是核心调用它请求堡垒机向目标机发起新的 TCP 连接返回的 channel 可以像 socket 一样recv和send。pump函数负责两个方向的数据搬运。4KB 的 buffer 是通用选择太小会放大系统调用开销太大会增加延迟4KB 在 SSH 转发场景里足够。两个线程分别处理socket - channel和channel - socket避免某一方向的数据把另一方向堵死。3.3 验证用 ssh 命令登录并确认目标机看到的来源 IP代码跑起来后另开一个终端执行下面的命令登录目标机ssh -p 2222 deploy127.0.0.1客户端连接的虽然是本机 2222 端口但经过转发实际到达的是目标机的 22 端口。登录成功后你在目标机执行who或查看 sshd 日志会发现连接来源 IP 是堡垒机地址而不是你本机地址。如果登录失败优先看脚本窗口有没有异常输出。常见问题是堡垒机账号权限不够或目标机的 sshd 配置拒绝了来自堡垒机的连接。这一步跑通说明转发链路没问题后面所有高级功能都可以在这个基础上叠加。另外注意这个版本录不到任何操作内容因为经过direct-tcpip的是加密后的 SSH 流量。想要审计必须进入下一章的会话代理模式。4. 从跳板到堡垒做一个会录屏的 SSH 会话代理4.1 架构反转Python 同时做 SSH 服务端与 SSH 客户端第 3 章是客户端直接和目标机建立 SSH 会话Python 脚本只搬运字节流。现在要做的是把这条链路从中间打断Python 脚本在自己监听的端口上扮演 SSH 服务端等用户ssh进来后再从脚本侧向目标机发起第二条 SSH 连接。这样用户先认证到你的 Python 服务认证通过后你在 Python 里完成对目标机的认证然后用户的 shell 请求转发到目标机。两段会话在 Python 进程内对接数据在这里是明文可以录屏。实现这个方案paramiko 需要同时用两套逻辑ServerInterface处理服务端认证和 channel 请求SSHClient处理到目标机的连接。代码量比转发版本大但每一块都是必要的尤其是ServerInterface的回调方法决定了哪些用户、哪些 channel 类型被允许。4.2 两段会话对接invoke_shell 与 pump 明文的代码下面这段是会话代理的核心骨架。监听 2222 端口收到连接后交给Transport进入服务端模式认证通过且用户请求 shell就向目标机建立一个同名 shell channel然后双向泵数据。import socket import threading import paramiko LISTEN_PORT 2222 HOST_KEY_FILE /var/lib/bastion/host_rsa TARGET_HOST 10.0.1.20 TARGET_USER deploy TARGET_KEY /home/deploy/.ssh/id_rsa def load_or_create_host_key(): try: return paramiko.RSAKey.from_private_key_file(HOST_KEY_FILE) except FileNotFoundError: key paramiko.RSAKey.generate(2048) key.write_private_key_file(HOST_KEY_FILE) return key class BastionServer(paramiko.ServerInterface): def __init__(self): self.event threading.Event() def check_allowed_auths(self, username): return publickey,password def check_auth_publickey(self, username, key): return paramiko.AUTH_SUCCESSFUL def check_auth_password(self, username, password): return paramiko.AUTH_SUCCESSFUL def check_channel_request(self, kind, chanid): if kind session: return paramiko.OPEN_SUCCEEDED return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED def check_channel_pty_request(self, channel, term, width, height, pixelwidth, pixelheight, modes): return True def check_channel_shell_request(self, channel): return True def bridge(src, dst, logfile): while True: try: data src.recv(4096) except Exception: break if not data: break dst.sendall(data) logfile.write(data) logfile.flush() def handle_connection(client_sock): host_key load_or_create_host_key() server_transport paramiko.Transport(client_sock) server_transport.add_server_key(host_key) server BastionServer() server_transport.set_server_mode() server_transport.start_server(serverserver) chan server_transport.accept(30) if chan is None: client_sock.close() return # 向目标机发起 SSH 连接并打开 shell target paramiko.SSHClient() target.set_missing_host_key_policy(paramiko.AutoAddPolicy()) target.connect( TARGET_HOST, usernameTARGET_USER, key_filenameTARGET_KEY, ) target_ch target.get_transport().open_session() target_ch.get_pty() target_ch.invoke_shell() logfile open(f/var/log/bastion/{chan.get_id()}.log, wb) threading.Thread(targetbridge, args(chan, target_ch, logfile), daemonTrue).start() threading.Thread(targetbridge, args(target_ch, chan, logfile), daemonTrue).start() def main(): host_key load_or_create_host_key() sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) sock.bind((0.0.0.0, LISTEN_PORT)) sock.listen(200) while True: client_sock, _ sock.accept() threading.Thread(targethandle_connection, args(client_sock,), daemonTrue).start() if __name__ __main__: main()load_or_create_host_key里有一个关键细节RSA key 必须落盘保存否则每次启动都生成新指纹客户端会出现 host key 冲突。BastionServer的回调方法里check_auth_publickey和check_auth_password目前都直接放行生产环境要在这里对接你的用户体系。target_ch.get_pty()必须在invoke_shell()之前调用它会向目标机请求一个伪终端目标机的 shell 才能正确处理全屏程序、vim 这类交互。bridge里的 logfile 同时写两个方向的数据既包含用户输入也包含目标机输出这就是录屏的原始素材。这套骨架的问题也很明显当前登录目标机用的是固定账号没有做用户映射。真实堡垒机需要根据登录用户、目标 IP 和权限策略决定用哪个账号登录甚至动态分发一次性凭据。4.3 录屏与审计字段存什么、怎么回放录屏文件只是字节流要让它有审计价值必须和会话元数据关联。我的做法是给每个会话生成一个 session_id作为录屏文件名同时在审计表里记录下面的字段。字段示例说明session_id8f3a2c9e会话唯一标识也是录屏文件名userops_zhang登录堡垒机的用户名source_ip192.168.10.8客户端来源 IPtarget_host10.0.1.20目标机地址target_port22目标端口start_time2025-06-01 10:00:00会话开始时间end_time2025-06-01 10:12:33会话结束时间log_path/var/log/bastion/8f3a2c9e.log录屏文件路径每条会话在开始时插入一条记录结束时更新 end_time。回放时读录屏文件按字节流喂给终端模拟器即可还原屏幕。注意录屏文件里包含终端的转义序列比如清屏、光标移动、颜色控制符直接当文本看全是乱码这是正常的。一个小建议不要边录边直接写最终文件。先写临时文件会话结束后统一 rename 到正式目录避免会话中途崩溃导致录屏文件损坏。4.4 并发边界线程模型能撑到多少上面这版实现每个会话至少占用两个线程一个负责用户到目标的泵一个负责目标到用户的泵。加上 accept 线程和每个连接的 handler 线程一百个在线会话就是两百多个线程。我实测过的感受是线程数量在五百以内Linux 上问题不大超过五百GIL 和上下文切换开销会明显拉高延迟。如果你的场景超过这个量级有两个出路加线程池上限比如信号量控制在 200 个同时在线多余的排队或者改用 asyncssh 重写把每会话的阻塞线程变成事件循环里的两个 task。5. 常见问题与避坑录屏空白、方向键乱码、空闲断线的 5 个现场5.1 host key 冲突每次重启指纹都变用户直接拒绝连接现象用户反馈“连不上报 host key 变了”你重启服务后所有客户端都出现类似的告警。原因开发初期最容易踩的坑。RSAKey.generate(2048)每次生成的都是新密钥服务端指纹自然每次都变。客户端会认为有人在中间冒充服务端直接拒绝连接。解决把 host key 持久化到磁盘。用第 4 章的load_or_create_host_key启动时检查文件存在就加载不存在才生成。上线前把指纹分发给用户之后就不会再变。5.2 录屏文件全是乱码回放没任何可读内容现象录屏文件能写但打开看全是\x00、\xff之类的字节或者回放时屏幕是空的。原因录屏开始得太早把 SSH 握手的二进制数据也写了进去。第 3 章的端口转发版本录下来的永远是密文因为数据过去时还没有经过你的应用层解密。只有第 4 章的会话代理模式在invoke_shell()成功返回之后再写 logfile录到的才是明文。解决启动录屏的时机放在两个 channel 都进入 shell 状态之后。可以先发一个空命令等手段确认 shell 就绪再开始写文件。回放时用终端模拟器渲染转义序列不要用普通文本编辑器看。5.3 空闲几分钟就断线后台上线的人全被踢现象用户挂着 ssh 不动五到十分钟后连接被服务端关闭重连才能继续操作。原因网络设备、云安全组或目标机的 sshd 配置里通常有 idle timeout长时间没有流量就回收连接。你的 Python 服务没有主动维持心跳自然被对端掐断。解决在 paramiko 的 transport 上设置 keepalive。server_transport.set_keepalive(30)每 30 秒发一个加密的 keepalive 包客户端侧也可以让用户配ServerAliveInterval 15两边都保活空闲断线问题基本消失。5.4 用户关掉终端会话状态一直不结束现象用户窗口一关审计记录里 end_time 迟迟不写在线会话数只增不减。原因pump 线程里只处理了单侧 EOF。用户断开时从 chan 读到空数据这个方向线程退出另一方向的 pump 还阻塞在target_ch.recv()上因为 channel 没有立刻收到 EOF线程就一直挂着。解决给 channel 设置超时比如chan.settimeout(1)recv 超时后检查对端状态和exit_status_ready()一旦任一侧 EOF 就主动关闭对方 channel并让两个线程退出。会话结束时统一更新 end_time。5.5 并发一高就卡在线人数一多延迟直线上升现象五十人以内正常两百人同时在线时操作明显卡顿CPU 不高但整体响应慢。原因每会话两个线程加上 GIL线程切换和 socket syscall 成为瓶颈。问题的典型特征是 CPU 使用率不高但延迟很高说明卡在调度和等待。解决用信号量限制同时处理的会话数比如设置 200 上限让请求排队而不是无限开线程。超过这个规模直接考虑用 asyncssh 重写核心链路的 pump 逻辑把线程模型改成事件循环模型。这个切换不是小工程但并发收益非常明显。6. 进阶技巧把 Python 堡垒机挂进 OpenSSH 的 ProxyCommand并给录屏加校验6.1 用户接入用 ProxyCommand 隐藏堡垒机地址给用户直接发ssh -p 2222 user127.0.0.1不够优雅而且要求每台客户机都记得端口和主机。更常见做法是给受管客户端分发一个 ssh config 片段用 ProxyCommand 调 Python 脚本完成接入Host prod-* ProxyCommand python3 /opt/bastion/proxy.py %h %p User deploy%h和%p由 OpenSSH 替换为用户执行 ssh 时传的主机名和端口。proxy.py 里连接堡垒机的服务端口打开到目标机的 channel再把进程的 stdin/stdout 与 channel 对接。这个方案的好处是用户无感还是正常 ssh代价是 ProxyCommand 模式下能做的是转发编排内容审计还得靠第 4 章的会话代理。6.2 给录屏加校验防篡改从会话结束那一刻开始录屏文件落盘后用 SHA-256 计算整个文件的摘要将会话 ID、文件路径、摘要写进审计库。回放工具打开文件时重新计算比对不一致就标记告警。生产环境我会再进一步每会话写文件时边写边算哈希会话结束后先记摘要再归档。使用上有些团队直接把摘要链成哈希链让任何一条记录的修改都会破坏后续所有记录取证时更有说服力。这个做法不复杂但把审计从“能看”升级成了“可信”。我自己最深的教训是早期那版只做了转发上线后被人拿着日志问“这操作是不是某人干的”我却拿不出任何内容证据。从那以后我把录屏和元数据落库放在了所有功能之前。堡垒机的价值不在于链路通不通而在于出了事你能不能拿出让人信服的记录。这套方案不完美但足够把链路、审计和回放串起来跑通剩下的权限粒度、账号映射、审批流都可以在这个骨架上慢慢长。希望帮到你。本文还有配套的精品资源点击获取