Python+Paramiko打造SSH堡垒机：从协议到会话审计实战

简介Python实现堡垒机的技术说明资源面向需要搭建安全运维审计系统的运维开发与安全测试人员聚焦如何利用Python生态完成远程访问控制、权限审计等核心能力。压缩包约58KB文件总数为0资源主体以文字描述为载体未附脚本或数据文件。内容较为系统地梳理了10个关键知识点包括基于socket与paramiko的SSH网络通信、用户名密码与公钥身份验证、RBAC权限管理、logging会话审计、threading与asyncio并发处理、Flask/Django Web管理界面、sqlite3/PostgreSQL数据持久化、失败登录限制与会话超时、REST API集成以及unittest/pytest测试与异常处理。对希望从零设计堡垒机原型的开发者而言这份资料可帮助快速明确技术选型和功能模块边界已有273人学习。1. Python 写堡垒机与其说是开发不如说是在搭一条看得见每一字节的通道做运维的人迟早会遇到一个坎某天研发负责人过来说新同事需要登一下生产环境的数据库跳板机你不想直接给 root 密码更不想让他在那台机器上敲rm -rf追悔莫及。最常见的解法是上堡垒机但商业堡垒机贵、开源堡垒机部署又重。于是很多人会问一句用 Python 自己写一个行不行答案是能行而且比你想的简单。所谓堡垒机本质上是一个放在用户和目标服务器之间的 SSH 代理层用户连的是你你再替他去连真正的机器。中间的流量、命令、会话、录像全部经过你的代码所以审计和控制都发生在这一层。这篇文章我会从 SSH 协议拆解讲到用 Paramiko 把最小可用的代理跑通再讲录像、审计和那几条你避不开的坑。适合有 Python 基础、想给团队做内部运维工具的从业者不需要你精通 TCP/IP但得愿意对着strace和ssh -vvv看输出。2. 堡垒机到底拆成几层从协议角度看清一条 SSH 链路2.1 用户、代理、目标机三者之间的两段 SSH 会话写堡垒机之前先得把链路画对。用户手里拿的是一台自己的笔记本目标机是机房里的某台 Linux 服务器。堡垒机代码跑在一台单独的跳板主机上它的工作方式是用户/带opsbastion_host建立第一段 SSH 连接这时候你的 Python 进程需要扮演 SSH Server 的角色你验证完用户身份之后再用自己的凭据向目标机发起第二段 SSH 连接扮演 SSH Client。两段会话在中间对接用户敲的每个字符经你转发到目标机目标机的每个回显也经你转回给用户。你可以把整个数据通路想象成两根管道对接Python 的paramiko库恰好同时具备 Server 端和 Client 端的能力所以两端接口都能落在同一套代码里。这里有一个容易被新手的直觉带偏的点很多人会想我直接在程序里os.system(ssh usertarget)不就好了不行那样你只能拿到一个子进程的 stdin/stdout拿不到 SSH 通道的精细控制权。常见做法是让 Paramiko 同时管理两个Channel对象一个来自服务端连接一个来自客户端连接然后在这两个 Channel 之间做字节级的搬运。Channel 是 SSH 协议里的逻辑通道终端、shell、SFTP 都跑在它上面。理解了这个你就知道堡垒机不是一个页面系统而是一个常驻的 TCP 监听程序。2.2 为什么选 Paramiko 而不是 AsyncSSH先聊并发模型再谈功能选型上我一般直接推 Paramiko。它不是性能最好的选项但 API 稳定、文档多、踩坑记录全网都是遇到问题搜得到答案。更重要的是Paramiko 的ServerInterface类把 SSH Server 该实现的钩子都列好了你只需要按需重写check_auth_*、check_channel_*这些方法。AsyncSSH 性能更好支持 asyncio但它的 Server 接口封装更重偏向协议库而不是代理工具。对一个内部堡垒机来说100 个并发会话已经是中等规模Paramiko 的阻塞模型配线程池完全吃得消。真到上千并发再考虑异步重构不迟那是另一个量级的问题。再说运行形态。堡垒机进程应该以守护进程方式跑在跳板机上随便用systemd或supervisor拉起来都行。我习惯的做法是代码里只负责监听和转发不保管用户、不保管资产用户表放在 SQLite 或 MySQL 里认证时现场查。这么拆的好处是以后想接公司的 LDAP 或钉钉扫码登录只改认证函数就行。下面这节我们就开始写代码。3. 用 Paramiko 跑通一个最小可用 SSH 代理3.1 写一个二十行的 ServerInterface让进程先学会装成一台 SSH ServerParamiko 的 Server 模式核心是继承paramiko.ServerInterface它定义了一堆check_*回调。你不需要全部实现只需实现你要用的认证方式和 channel 类型。下面这段代码我写的时候把登录校验放在了check_auth_password里真实项目里这里应该查数据库或 LDAPdemo 就先用常量。import paramiko class BastionServer(paramiko.ServerInterface): def __init__(self, client_ip): self.event threading.Event() self.client_ip client_ip def check_channel_request(self, kind, chanid): # 只允许申请 session 类型的通道禁止直接转 channel if kind session: return paramiko.OPEN_SUCCEEDED return paramiko.OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED def check_auth_password(self, username, password): # demo 用固定账号生产环境改成查用户表或 LDAP if username ops and password ops2024: return paramiko.AUTH_SUCCESSFUL return paramiko.AUTH_FAILED def check_channel_pty_request(self, channel, term, width, height, pixelwidth, pixelheight, modes): # 允许分配 PTY这是交互式终端的前提 return True def check_channel_shell_request(self, channel): return True def get_banner(self): return (bBastion ready, please login\r\n, en-US)这段代码的逻辑并不复杂check_channel_request决定是否接受对方发起的通道申请这里只放行session否则后续压根走不到 shell。check_auth_password做用户名密码校验返回AUTH_SUCCESSFUL或AUTH_FAILED。check_channel_pty_request必须返回True否则用户拿到的是没有终端能力的通道vim、top 这类程序会直接报TERM错误。get_banner里返回的字符串会在用户登录前显示在 SSH 客户端上适合放一句告警。这里有个细节你可能会忽略check_auth_password是同步阻塞的如果你在里面查数据库遇到慢查询会让整个 SSH 握手卡住。我一般会把认证逻辑丢进线程池用concurrent.futures包一层数据库慢不拖垮握手。3.2 双向转发与连接管理两条线程一个都不能省ServerInterface 只是协议侧的壳真正的数据流在handle_connection函数里。当 SSH 握手完成、用户发出 shell 请求后Paramiko 会通过Transport对象的accept返回一个Channel对象这个 Channel 代表用户侧。紧接着你要用paramiko.SSHClient发起第二段连接拿到目标侧的 Channel。然后让两个 Channel 互相对发。代码如下import threading import paramiko def forward_channel(src_channel, dst_channel): 把 src_channel 收到的数据原样写到 dst_channel直到 EOF。 try: while True: data src_channel.recv(32768) if not data: break dst_channel.send(data) except (EOFError, OSError) as e: pass finally: # 一个方向断了说明会话基本结束尽力关掉另一端 try: dst_channel.close() except Exception: pass def handle_connection(client_sock, client_ip): transport paramiko.Transport(client_sock) transport.add_server_key(host_key) server BastionServer(client_ip) transport.start_server(serverserver) chan transport.accept(20) if chan is None: transport.close() return # 第二段连接用代理的凭据去连真实目标机 target paramiko.SSHClient() target.set_missing_host_key_policy(paramiko.AutoAddPolicy()) target.connect(10.0.0.50, usernamedeploy, key_filename/etc/bastion/deploy_key) target_chan target.get_transport().open_session() target_chan.get_pty(termchan.get_pty_terms(), widthchan.get_pty_width(), heightchan.get_pty_height()) target_chan.invoke_shell() # 双向转发各用一条线程防止一端阻塞影响另一端 t1 threading.Thread(targetforward_channel, args(chan, target_chan)) t2 threading.Thread(targetforward_channel, args(target_chan, chan)) t1.start() t2.start() t1.join() t2.join() target.close() transport.close()两个方向各起一条线程是必须的别图省事用一个循环轮询因为recv是阻塞的一个方向没数据会在recv里挂起另一个方向的数据就没人转了。recv(32768)的缓冲区大小按网卡 MTU 折中来的改成 4096 也能跑但大文件传输时小缓冲区会导致用户侧飘红的速度上不去。target_chan.get_pty(...)里的三个参数是从用户侧 Channel 上取来的终端类型和窗口尺寸目标机必须用同样的 PTY 参数创建会话否则用户的 vim 界面和实际渲染宽度对不上。这里有个容易翻车的地方target.connect是同步的如果目标机网络不通默认会卡上几十秒。建议在connect里加timeout10会话建立失败时就给用户返回一行明确报错而不是让他在黑屏前等。3.3 参数怎么设关联并发、超时与登录校验的取舍最小实现跑通之后参数就要开始较真了。先说监听线程数。我上面用的是每来一个 TCP 连接就起一个handle_connection线程这在线程开销可接受的前提下最简单。但这个架构顶不住上千连接线程数会失控。常见做法是外面套一个固定大小的ThreadPoolExecutor最大并发设成 CPU 核数乘 8超出的连接先让它们在 socket 层排队。超时参数要分三层设transport.accept(20)是等待用户发起 channel 申请的超时超时返回None就直接关闭target.connect的timeout是 TCP 连接超时还有一层是会话空闲超时用户挂着不动太久应该被踢下线。Paramiko 没有内置的空闲检测我一般会在forward_channel里维护一个last_activity时间戳用select的timeout参数做轮询超过session_idle_limit就主动断开。登录校验这块要聊一下安全性。check_auth_password暴露的是明文校验接口Paramiko 在握手阶段已经把密码加密传输了所以不担心抓包但你要防的是暴力猜解。接了个 MySQL 后每次失败都往失败表里写一条连续失败五次锁 IP 十分钟这个逻辑用 SQLite 就能轻松实现。资产关联不要写死在handle_connection里我的习惯是用户登录成功之后给一个菜单按数字键选目标机选完再去建立第二段连接。菜单本身也是一个invoke_shell只是把用户输入映射成目标机器列表的下标。3.4 最小版的运行验证用ssh -p 2200走通全链路代码写完先别急着接数据库直接把监听端口起在本地 2200用系统自带的ssh客户端连一把。# 生成一个测试用 host key不要用默认的防止连公共密钥都撞车 ssh-keygen -t ed25519 -f /etc/bastion/ssh_host_ed25519_key -N # 启动堡垒机服务后台运行 nohup python3 bastion.py /var/log/bastion/access.log 21 # 本地连一把试试 ssh -p 2200 ops127.0.0.1 # 如果登录后黑屏或卡住加大 verbosity 重新连 ssh -vvv -p 2200 ops127.0.0.1ssh -vvv的输出是这条链路最好的排障工具。你会在上面看到Authentications that can continue: password、Channel: session requested这些关键字。如果停在Authentications that can continue不前那就是check_auth_password没返回AUTH_SUCCESSFUL如果停在shell request accepted那就是check_channel_shell_request没写对。验证通过后把监听端口改成 22再配 systemd 拉起一个能用的 Python 堡垒机骨架就成型了。4. 会话审计与录像堡垒机的价值不在打通而在留痕4.1 录像文件怎么组织原始流和清洗流分开存代理层建好之后审计就是堡垒机能不能在出事时自证的关键。最原始的审计资产是双向转发的每一字节数据。很多人第一反应是直接把这些字节存成文件然后发现回放时满屏都是[38;5;196m这类颜色转义序列根本没法看。常见做法是原始流和清洗后的展示流分两个文件存原始流按时间戳追加到一个二进制日志清洗流是同一份数据经过 ANSI 过滤器之后的可读文本。import re import time ANSI_PATTERN re.compile(rb\x1B\[[0-9;?]*[ -/]*[-~]) class SessionRecorder: def __init__(self, session_id, path): self.session_id session_id self.raw_path f{path}/{session_id}.raw.log self.clear_path f{path}/{session_id}.clear.log self.raw_fp open(self.raw_path, ab) self.clear_fp open(self.clear_path, ab) def write(self, data): ts int(time.time() * 1000) # 原始流带毫秒时间戳前缀方便回放对齐 self.raw_fp.write(f{ts}:.encode() data b\n) # 清洗流去掉 ANSI 控制符只留可读文本 clean ANSI_PATTERN.sub(b, data) self.clear_fp.write(clean) self.clear_fp.flush()录像文件按会话 id 分文件存文件名就是会话的唯一标识。raw文件每一行加毫秒时间戳但格式上有一个坑SSH 字节流里本来就可能包含换行和多行文本直接把时间戳和数据拼到一行回放时没法精确切分。我现在的做法是把时间戳作为每行前缀同一批数据里换行符替换成\r\n序列回放解析时先按时间戳切段再还原。这个格式自己定就行但要保证一个原则原始流不能丢字节回放进度的每一秒都能对应到当时的屏幕渲染。4.2 命令级审计把用户敲的每一条命令挖出来落库录像文件是会话级的但审计时往往想直接查到「谁在哪个时间点执行了rm -rf /data」。这就得做命令级提取。最土但最有效的办法是监听 shell 的提示符因为用户敲完命令回车后shell 会把回声和输出一起回传提示符出现的位置就是一条命令的结束。下面这段提取逻辑放在write方法里PROMPT_RE re.compile(rb\$|#|) def extract_commands(self, data): 从清洗流里按提示符切出用户输入的完整命令。 self.line_buffer data commands [] while True: match PROMPT_RE.search(self.line_buffer) if not match: break # 提示符之后到下一个提示符之前是用户的输入 seg self.line_buffer[match.end():] next_prompt PROMPT_RE.search(seg) if next_prompt: cmd seg[:next_prompt.start()].strip() if cmd: commands.append(cmd.decode(utf-8, errorsreplace)) self.line_buffer seg[next_prompt.start():] else: break return commands这里要注意errorsreplace用户敲中文命令或终端传回 UTF-8 之外字节时解码失败会把整条命令丢掉。实际项目里我会把这条命令连同会话 id、时间戳一起插进审计表并挂一个cmd_hash用来做敏感命令匹配。命令级审计只是辅助不要用清洗后的文本替代原始录像因为清洗过程一旦有 bug审计证据就哑了。我见过有人在正则上踩坑把包含ESC的合法内容也吞了最后回放和命令表对不上。4.3 权限与资产表设计代理层不该做成全通门能记录之后就该管权限了。堡垒机的核心价值之一是「用户能看到哪些机器、登录后以什么身份执行命令」。我的资产表通常会建三张表users、assets、user_asset_perms。用户表不存密码存的是公钥和认证方式标记资产表存目标 IP、端口、登录账号权限表把用户和资产关联并带一个can_sudo字段。CREATE TABLE users ( id INTEGER PRIMARY KEY, username TEXT NOT NULL UNIQUE, public_key TEXT NOT NULL, enabled INTEGER DEFAULT 1 ); CREATE TABLE assets ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, host TEXT NOT NULL, port INTEGER NOT NULL DEFAULT 22, remote_user TEXT NOT NULL ); CREATE TABLE user_asset_perms ( id INTEGER PRIMARY KEY, user_id INTEGER NOT NULL REFERENCES users(id), asset_id INTEGER NOT NULL REFERENCES assets(id), can_sudo INTEGER DEFAULT 0 );登录流程变成用户连上来后认证通过程序根据用户名查出他有权限的资产列表动态生成一个菜单。他选资产时才发起第二段连接。这样的话目标机的 IP 和端口不会暴露给用户随心所欲拼 IP 也连不上不该连的机器。这里要特别提醒remote_user不要用 root目标机上创建一个普通运维账号需要提权时让堡垒机在invoke_shell后发送sudo -i并把can_sudo字段作为是否发送该命令的判断依据。这样能避免用户拿到 root shell 直接绕过堡垒机。5. 堡垒机避坑与常见问题排查5 条血泪经验5.1 坑一recv返回的字节数不稳定转发循环直接卡死现象用户连上后一切正常但运行一段时间后终端突然挂起踢掉重连又好了。检查代码发现转发循环并没有退出就是卡在channel.recv上。原因Paramiko 的Channel.recv在没有数据时会阻塞而另一端的连接已经半关闭此时recv可能返回空字节串而不抛异常。如果代码只判断if not data: break空串被误判为 EOF通道被关闭但如果数据还在半路上你就永远不会走进break。更隐蔽的是大流量时两个转发线程的调度竞争导致一个线程长期拿不到 GIL另一个线程的send缓冲区写满全家卡死。解决给recv设置超时配合空串统计判断def forward_channel(src, dst): src.settimeout(5) empty_count 0 while empty_count 3: try: data src.recv(32768) if not data: empty_count 1 time.sleep(0.1) continue empty_count 0 dst.send(data) except socket.timeout: empty_count 1 continue except Exception: break dst.close()这个逻辑把「阻塞等待」改成「限时等待」连续三次拿不到数据才判定对面挂了。代价是每秒多几次系统调用对内部工具来说可忽略。5.2 坑二窗口尺寸变化不转发vim/htop 界面乱成一团现象用户在本地终端把窗口拉大或缩小堡垒机里的 vim 界面却保持初始行列数内容要么截断要么折行错乱。原因SSH 协议里窗口尺寸是通过pty-req消息带的之后窗口变化还需要window-change消息同步。我第一版代码只在建立会话时设置了初始尺寸完全没有监听之后的尺寸变化。解决在前端连接里捕获窗口变化并调用resize_pty同步到目标通道。Paramiko 的 Channel 有一个resize_pty(width, height)方法你只需要在用户侧 Channel 上读出新尺寸再对target_chan调用一次。在handle_connection里往forward_channel的循环之前额外起一个专门监听尺寸变化的线程。def sync_window_size(chan, target_chan): while True: width chan.get_pty_width() height chan.get_pty_height() if (width, height) ! target_chan.get_pty_width(), target_chan.get_pty_height(): target_chan.resize_pty(widthwidth, heightheight) time.sleep(0.3)5.3 坑三录像只存清洗流审计回放时证据成哑帖现象负责安全合规的同事某天要求导出某时段会话记录你导出的清洗流里命令是有的但用户当时看到的界面颜色、加粗警告、vim 的分屏布局全没了回放还原度很差。原因清洗流是给文本检索用的不是给回放用的。ANSI 过滤器在去掉控制符的同时把终端的渲染状态机也扔了。解决原始流必须完整保存清洗流只作为检索索引。回放工具直接用原始流驱动终端模拟器清洗流仅用于命令提取和全文搜索。任何情况下不要用一个文件同时承担两种职责。5.4 坑四close 的顺序不对远程进程变孤儿现象用户正常退出终端后目标机上由该会话启动的tail -f或top进程还挂着下次登录发现一堆残留进程。原因先关transport.close()会导致底层 TCP 直接 RST目标机收到的是连接异常终止而不是正常 EOF目标 shell 收不到 SIGHUP子进程继续存活。解决关闭顺序应当是先close两侧的 Channel再close传输层。我在forward_channel的finally块里做了标准关闭但要确保两个转发线程都退出后再关transport。主流程里用t1.join()和t2.join()等两条线程结束然后才执行transport.close()。如果用户断线但目标机还有进程还要再发一个channel.close()尝试触发目标侧 hangup。5.5 坑五SSH 客户端用 RSA 密钥连不上日志里写的是算法不支持现象用户在本地ssh-keygen生成了一个默认 RSA 密钥填好公钥后连接堡垒机客户端直接报server does not support public key algorithm ssh-rsa或类似的 verbose 输出密码登录却正常。原因Paramiko 3.x 默认禁用了ssh-rsaSHA-1 签名算法而很多旧工具生成的默认 RSA 密钥用的还是 SHA-1 签名。你检查check_auth_publickey并没有写错纯粹是算法协商阶段就没达成一致。解决两选一。让用户换成ssh-keygen -t ed25519生成密钥或在你启动 Transport 之前显式启用 SHA-1 算法。前者是推荐方式后者只适用于公司内部实在无法换密钥的场景且要在代码里额外配置算法列表。注意如果用户用的是系统自带 SSH 客户端它每次默认生成的密钥类型会随版本变化建议在文档里统一写上ed25519省得排查半天。6. 从 SSH 裸代理到生产可用验证方法、压测与 Web 化扩展先给压测量化一下你的代理层能扛多大并发。别用 JMeter 这种 GUI 工具直接 shell 循环开连接就够了# 并发 100 个连接每个执行 exit 后断开统计失败数 for i in $(seq 1 100); do (ssh -p 2200 -o ConnectTimeout5 -o StrictHostKeyCheckingno \ ops127.0.0.1 echo ok /dev/null 21 \ echo success /tmp/bastion_press.txt || \ echo fail /tmp/bastion_press.txt) done wait grep -c success /tmp/bastion_press.txt如果一批 100 个连接里失败率超过 5%大概率不是带宽问题而是你的线程池或者 Paramiko 的Transport握手线程没有及时清理进程的 fd 数涨到了 ulimit 上限。用ss -s看 socket 状态如果大量TIME_WAIT堆积调整系统net.ipv4.tcp_tw_reuse参数能缓解。这一步务必在接入真实用户前做否则上线第一天就会被群聊里的「连不上」刷屏。再往后走常见的进阶路线是把 SSH 裸代理改成 Web 化堡垒机。前端用 Xterm.js 渲染终端后端把 WebSocket 收到的数据转成对 Paramiko Channel 的读写。这个改造不替换核心逻辑只需在forward_channel外面包一层 WebSocket 的send/receive适配器录像和审计能原样复用。这样用户不用装 SSH 客户端浏览器打开就能干活。但代价是要处理 WebSocket 的 ping/pong 心跳、断线重连、多标签页会话冲突每一个都是新的坑。我的习惯是新方案先让一小撮人试用两周只开放非生产资产每天手动抽查录像文件和命令审计表确认没有漏记、错记才逐步扩大范围。堡垒机这种工具出一次信任危机就很难再让人安心用宁可上线慢一点。最后提醒一句host key 一定要定期轮换别因为没人连接就不管密钥泄露的后果比你想的严重。希望帮到你。本文还有配套的精品资源点击获取