简介这是一份面向有一定Python基础的开发者的堡垒机安全运维审计系统实现指南聚焦远程访问的集中管理、用户权限控制与操作审计等企业级安全运维需求。资源包约58KB内容精炼已有273人浏览学习。资料系统梳理了使用Python构建堡垒机所需掌握的关键技术基于socket与paramiko库实现加密通信和远程登录利用getpass安全采集用户密码设计RBAC角色权限模型借助logging完成全量会话审计同时介绍threading与asyncio两种方式处理并发请求结合Flask/Django搭建可视化Web管理界面并通过sqlite3、PostgreSQL等数据库持久化存储用户、权限与日志信息。此外还涉及设置登录失败限制、会话超时等安全策略以及RESTful API集成和unittest/pytest自动化测试方法帮助开发者从架构设计到编码落地形成一套完整、可扩展且符合审计要求的堡垒机开发方案。1. python 写堡垒机先想清楚要转什么、审什么、控什么去年帮某团队处理过一次丢审计记录的事故线上误删数据翻历史命令记录终端里只剩下一点零散残留时间线完全对不上。那之后我才把「python 写堡垒机」从业余项目提成了正经需求。这里的堡垒机不是商业产品而是基于 Paramiko 这类 SSH 协议库自己搭一套「用户终端 → 堡垒机 → 目标机」的三层中转层把录像回放、命令审计、权限黑白名单全部做进同一套代码里。适合中小团队运维机器多、要对接内部工单系统、又不想被商用堡垒机协议绑死的场景这套思路比通读一遍源码再改二次开发要省事得多。下面按我实际拆过和排过障的路线讲侧重能直接抄的代码和气死人的坑。2. 核心架构SSH 会话三层链路与 Paramiko 的两种打开方式在动手写第一个转发循环之前先把堡垒机在 SSH 层面的结构讲清楚。绝大多数人第一次写堡垒机的时候都低估了「中转」这两个字的复杂度它不只是多一次 ssh 登录而是要同时维护两套 channel并在中间做数据搬运和审计插桩。2.1 堡垒机的工作链路三层会话与两条数据流从网络拓扑上看堡垒机串在用户与目标机之间用户终端 --SSH-- 堡垒机 --SSH-- 目标机用户本地的 ssh 客户端连的是堡垒机堡垒机内部再通过 Paramiko 往目标机发起第二条 SSH 连接。两条连接各自有独立的 channel真正的命令回显先到堡垒机再被转发回用户终端。中间这一段「搬运」就是堡垒机全部价值所在它能看到所有输入输出能记录时间戳能对输入做危险命令拦截。在具体代码上paramiko 打开远程 shell 有两种方式exec_command 和 invoke_shell。这两者的差异直接决定你的堡垒机是「能跑」还是「能用」。打开方式适用场景是否分配伪终端是否支持交互式命令exec_command一次性执行命令、批量采集否非交互不支持 vim/top/viinvoke_shell get_pty堡垒机人工登录场景是Shell 环境完整支持交互式命令我一般会强制要求堡垒机场景里用 invoke_shell get_pty。原因很朴素运维人员会敲 vi、看 top、进 python 解释器这些操作离开 PTY 直接退化。exec_command 只适合监控类或巡检类的非交互任务混用会出现「用户进不了数据库客户端」「top 显示乱码」这类问题。这里有一个常见误区很多初版代码喜欢把 exec_command 当默认选项因为它的返回流好读适合做 pytest 测试。但如果你的堡垒机用户要操作真实环境exec_command 拿到的数据流没有终端控制序列连 top 的刷新都收不到。判断标准很简单用户会不会在会话里按方向键、CtrlC、Tab 补全会就上 invoke_shell。提示get_pty 之后建议显式声明终端类型和尺寸。默认的 term 是 vt100很多现代命令的界面会错位。2.2 最小可用实现Paramiko 接管 SSH 会话与伪终端分配直接给一段能跑的最小堡垒机核心代码包含用户认证和目标机连接两个环节import paramiko import threading import socket def connect_target(host, port, username, password): ssh paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) # 这里把超时参数一次性设置齐全 ssh.connect(host, portport, usernameusername, passwordpassword, timeout10, banner_timeout15, auth_timeout15) return ssh def open_shell(ssh): channel ssh.invoke_shell(termxterm) # 分配伪终端并指定窗口尺寸防止 tab 自动补全或 top 排版错乱 channel.get_pty(termxterm, width160, height48) channel.setblocking(False) return channel这段是后面所有功能的地基。set_missing_host_key_policy(AutoAddPolicy) 用于自动接受目标机 host key内网场景够用如果合规要求严格改成 WarningPolicy 并记录异常指纹更稳妥。timeout10 是 TCP 建连超时banner_timeout 指 SSH 协议握手时等待 banner 的时间auth_timeout 是密码认证耗时上限。这三个参数分开设置的意义在于目标机不通时TCP 层快速失败协议层卡住时banner 超时兜底密码验签卡住时auth 超时兜底不会让用户无限等待。setblocking(False) 是后续能写并发的前提。阻塞模式下channel.recv() 会一直占着当前线程一旦有几个并发连接线程池直接被打满。非阻塞后每次 recv 拿不到数据就返回空字节配合 select 或 sleep 做轮询即可。2.3 数据搬运与并发从阻塞转发到多会话隔离有了两端的 channel 之后剩下就是把目标机的输出搬回用户终端把用户的输入搬到目标机。这里要小心一个经典翻车点直接把 recv 到的字节原样 send不做分片和缓冲高延迟网络下会出现「敲一个字符卡一次」。import select import time import threading def forward(channel_c, channel_d, audit_logger, idle_timeout300): last_active time.time() while True: ready_r, _, _ select.select([channel_c, channel_d], [], [], 0.1) if channel_c in ready_r: data channel_c.recv(4096) if data: channel_d.sendall(data) audit_logger.record_input(data, time.time()) last_active time.time() if channel_d in ready_r: data channel_d.recv(4096) if data: channel_c.sendall(data) audit_logger.record_output(data, time.time()) last_active time.time() if time.time() - last_active idle_timeout: channel_c.close() break if channel_c.closed or channel_d.closed: breakselect.select 监听两个 channel 的可读状态0.1 秒的超时保证循环不会空转。recv(4096) 是分片大小可以根据屏幕输出量调整一般 4096 字节足够。idle_timeout 用来回收闲置会话运维环境里默认 300 秒比较合理太长占着连接资源太短容易把长任务误杀。注意 last_active 只在真正收到数据时更新如果用户长时间挂机没有输入这个会话会被自动回收。到这里一个单会话堡垒机已经能跑通。接下来是并发。paramiko 的 SSHClient 不是设计成全局对象复用的它内部有独立的 transport 线程错误地共享一个 client 会导致通道混乱。常见做法是每个用户会话独立创建 SSHClient 和 channel放在独立线程里处理。def handle_user(user_channel, user_auth, target_info): # user_channel 是用户连到堡垒机的 channel target_ssh connect_target(target_info[host], target_info[port], target_info[username], target_info[password]) target_channel open_shell(target_ssh) audit AuditLogger(session_idtarget_info[session_id]) forward(user_channel, target_channel, audit) # 会话结束后关闭两端连接 target_channel.close() user_channel.close()这里每个 handle_user 跑在一个线程里线程之间互相不共享任何参数字段。session_id 要在最外层生成后面录像文件、审计表都以它为主键。线程模型在超过两百并发时会吃力但那已经不是 Python Paramiko 该面对的场景了真到那个规模建议上异步重写或直接换商业方案。3. 审计落地录像回放、命令切片与结构化存储堡垒机的核心竞争力不是「能跳转」是「能审计」。这一章把审计链路拆成三层原始录像、命令切片、结构化存储。这三层缺一层等出事故翻记录的时候都会发现根本不够用。3.1 会话录像抓输入输出流并打时间戳录像的本质是记录用户在什么时间输入了什么、看到了什么。很多人第一版只记输出流误删数据后回放只能看到命令执行结果看不到用户到底敲了什么——这等于白记。import os import time import json class AuditLogger: def __init__(self, session_id, record_dir./records): self.session_id session_id self.input_f open(os.path.join(record_dir, f{session_id}.input), wb) self.output_f open(os.path.join(record_dir, f{session_id}.output), wb) self.meta_f open(os.path.join(record_dir, f{session_id}.meta), w) def record_input(self, data, ts): self.input_f.write(data) self.input_f.flush() def record_output(self, data, ts): self.output_f.write(data) self.output_f.flush() def close(self): self.input_f.close() self.output_f.close()input 文件存用户按键原始字节output 文件存目标机回显meta 文件在会话结束时写入用户、目标机、开始结束时间。三个文件用同一 session_id 归档。为什么 input 和 output 分开回放时能区分「用户敲了什么」和「系统显示了什么」对命令误操作归因非常关键。多数黑匣子问题就出在只存了一份合并流回溯时根本分不清是用户主动输入还是程序输出。回放格式上直接复用 Linux script 命令的 typescript 时间戳格式是省事的选择每行前加一个时间偏移量回放时按偏移量 sleep 后逐行刷出。这个格式 Vim 的 vimlog 插件、expect 脚本都能读不用自己造播放器。3.2 命令级审计把交互式输入拆成一条条可读命令做过终端抓包的人都知道用户在交互式 Shell 里敲的不是一行行命令而是一个个字节。用户敲 cd /tmp 再按回车input 流里是 bcd /tmp\r中间还掺杂退格符、方向键序列、Tab 补全的高亮代码。直接按换行 split 出来的记录全是乱的。我的做法是在审计器前面接一个行缓冲器先把 ANSI 控制序列和退格吃掉再按回车拆分命令。import re class CommandSplitter: def __init__(self): self.buffer self.ansi_re re.compile(r\x1b\[[0-9;?]*[a-zA-Z]) def feed(self, data: bytes) - list: text data.decode(utf-8, errorsreplace) text self.ansi_re.sub(, text) # 去掉终端控制序列 text text.replace(\x7f, ) # 去掉 DEL 退格 text text.replace(\r\x1b[K, ) # 去掉行清屏序列 self.buffer text commands [] while \r in self.buffer or \n in self.buffer: idx min( [i for i in [self.buffer.find(\r), self.buffer.find(\n)] if i 0] ) line self.buffer[:idx] self.buffer self.buffer[idx1:] if line.strip(): commands.append(line.strip()) return commandsCommandSplitter 的输出会以列表形式追加进审计日志。要注意 \r 和 \n 出现的顺序对拆分结果有影响Windows 风格的回车换行会连续触发两次所以拆完后要 strip 再记。这个方案只对普通 Shell 命令可靠进了 vim 或者 less 之后控制序列的拆分会变复杂。我的习惯是进交互子程序之后只记原始流出子程序再恢复命令级审计。判别方式是用正则匹配 vim/less/top 这类进程启动提示符。3.3 审计数据归档与回放接口录像文件攒多了会面临一个问题按什么命名、怎么快速找到某次事故的记录。我设计了一套归档规范三张表直接对应文件系统的三个目录。目录文件命名内容保留策略records/input{session_id}.input用户按键字节流按月度清理records/output{session_id}.output目标机回显字节流按月度清理records/meta{session_id}.meta会话元数据 json至少保存一年index/audit.db—command 表 session 表长期保留元数据记录用 json 一行写进去回放时先读 meta 拿时间偏移再按偏移读 input/output 两个二进制文件。一个小时的高强度操作会话input 加 output 通常在几十 MB 量级磁盘压力在可控范围内。回放接口不推荐做成 Web 播放器成本高还容易出兼容问题。我的习惯是暴露一个命令行工具输入 session_id 直接按 script 的格式回放。带时间戳的原始数据永远比花哨的网页可靠等真正需要给非技术人员看回放时再写一层转换成视频的脚本也不迟。4. 权限与资产用配置驱动一个可扩展的 RBAC 雏形没权限控制的堡垒机就是裸奔。真实的堡垒机还要求「用户能看到哪些资产、上去能执行哪些命令、不能执行哪些命令」这就是 RBAC 的核心。资源量在百台以内时不建议一上来就接数据库配置驱动的方式更快也更直观。4.1 资产与角色JSON 起步必要时换库一段典型的资产配置文件长这样{ users: { alice: { roles: [ops], ip_whitelist: [10.1.0.0/16], password_hash: $6$... } }, roles: { ops: { assets: [web-01, db-01], commands: {allow: [], deny: [reboot, shutdown]} } }, assets: { web-01: {host: 10.2.0.11, port: 22, username: root}, db-01: {host: 10.2.0.35, port: 22, username: opsadmin} } }这份 JSON 表达了三件事用户挂在角色下角色声明可访问的资产清单和命令限制资产声明连接参数。直接改 JSON 就能加人、加机器不需要重启服务——只要实现热加载。资产超过三百台的时候JSON 文件会变得难以维护再考虑迁到 MySQL但表结构设计基本照抄这份 JSON 的层级关系即可。这里有个大家容易忽略的点每个用户的连接参数不应该写在资产里。资产和用户之间要做一层账号映射用户 alice 在 web-01 上可能不是 root而是 opsadmin 或带 sudo 的普通账号。连接参数放在 roles.assets 里不太够用我会在 assets 下加 target_user 字段并在授权时强制检查该字段是否存在。少了这层审计记录里会出现一堆 root 账号合规检查直接翻车。4.2 访问控制黑白名单、IP 约束与超时回收黑名单是每个堡垒机的最低配置但这里有个执行层面的坑光拦命令前缀不够要拦参数。比如 deny [rm] 拦不住 rm -rf /因为命令名是 rm参数是 -rf /前缀匹配根本照不住。做黑名单匹配时要把完整命令行拆成命令名 参数列表再对参数做模式匹配。import re import shlex class CommandGuard: def __init__(self, deny_rules): self.deny_rules deny_rules def check(self, command_line): try: parts shlex.split(command_line) except ValueError: return False if not parts: return False cmd parts[0] joined .join(parts) for rule in self.deny_rules: if rule[type] exact and cmd rule.get(cmd): return False if rule[type] regex and re.search(rule.get(pattern), joined): return False if rule[type] param and cmd rule.get(cmd) and rule.get(params): if any(re.search(p, arg) for p in rule[params] for arg in parts[1:]): return False return True我在这类项目里的默认规则至少包含shutdown、reboot、mkfs、dd if、rm -rf / 和 drop database。shlex.split 会把引号里的空格正确处理掉避免用户拼一个带引号的命令绕过检测。param 型的规则要特别注意rm -rf / 这种参数是 /用正则要写 ^/$ 或者 ^/$避免把 rm -rf /tmp/xxx 也误拦。超时策略分两类绝对时长和空闲时长。绝对时长用来兜底通常设 12 小时强制断开会话并归档空闲时长用上一章的 idle_timeout默认 300 秒。这里有个经验长任务编译、数据迁移用户的空闲时间其实很短因为任务输出会持续刷屏所以 idle_timeout 按最后收到数据的时间算基本不会误杀长任务。4.3 登录到连接的完整时序把上面两节串起来一次完整操作请求链路是这样的用户通过 SSH 连接堡垒机完成登录认证密码或公钥。堡垒机读取用户角色和资产列表返回可见资产菜单。用户选择一个资产堡垒机执行授权检查IP 是否在白名单、资产是否属于角色。建立堡垒机到目标机的 SSH 连接分配 PTY启动 forward。forward 循环里挂上 CommandGuard每条输入命令先过黑名单命中则中断会话并记录告警。用户断开或超时会话关闭录像文件落盘审计索引写入。其中第 3 步的授权检查是很多人会漏掉的一环。我看到过一些精简版堡垒机选资产之后直接连目标机完全不校验资产归属。风险在于用户 A 只要知道目标机 IP就能绕过白名单直连。所以授权检查必须放在资产选择之后、建立连接之前且必须用配置文件里的 host 和端口而不是用户传入的任意地址。def authorize(user, target_id, user_ip): role_ids config[users][user].get(roles, []) allowed_assets [] for rid in role_ids: allowed_assets.extend(config[roles][rid].get(assets, [])) if target_id not in allowed_assets: return False, asset not in role whitelist config[users][user].get(ip_whitelist, []) if not ip_in_any_networks(user_ip, whitelist): return False, ip not allowed return True, ip_in_any_networks 可以用 ipaddress 模块实现按 CIDR 判断支持 10.0.0.0/8 这类网段。用户 IP 在登录握手阶段就能拿到SSH 的 peer 地址从 transport 里取代码里不要直接用 server-side 的 REMOTE_ADDR 概念免得与反向代理混淆。5. 避坑手记五个高频问题从现象到根因的排查记录这一章全部来自我实际调过的排障记录。每条按「现象 → 原因 → 解决」展开专治那些一上线就四处冒烟的小毛病。5.1 录像文件打开只有黑屏堡垒机 CPU 却接近 100%现象会话结束后用播放器打开录像文件画面完全黑屏但文件大小并不小。同时堡垒机 CPU 占用率异常高。原因forward 循环里没有事件让出机制。第一次写 forward 时直接 while True recv()recv 返回空后立刻进入下一轮循环等于一个死循环跑满单核。CPU 满载导致用户输入输出处理延迟录像落盘也变慢最终截图出来就是黑屏。解决在循环里加 select 或 sleep。我一般用 select.select 带 0.1 秒超时轮询两个 channel 的可读状态。CPU 占用能从 100% 降到 5% 左右。另外recv 之后要 flush 磁盘缓冲老版本代码只在会话结束时统一写文件中途进程崩了就丢全部录像。5.2 用户按键延迟严重敲一个字符半秒才回显现象用户连上堡垒机操作目标机每个键按下到回显出现有明显的延迟网络状况良好但延迟依然存在。原因数据搬运循环对 4096 字节的分片大小没做细分。用户输入通常一次只有 13 字节而回显往往几百字节一次。如果 recv 和 send 处理顺序不当回显会被积压在 recv 缓冲区里直到下一次输入才触发搬送。解决在内层做一个小批量循环同一轮 select 里反复 recv 直到返回空再统一 sendall。或者把 recv 改成 channel.get_pty 之后的阻塞模式但加大超时。经验值是数据交换频率越高、单次数据量越小越要用小分片加高频轮询的组合。5.3 命令审计记录出现 ^H 和方向键乱码现象审计表里出现 b^H、b^[[A 这类字符一条 cd 命令被拆成三条记录。原因直接把 input 流按 \r 拆分没有处理退格、方向键和 ANSI 控制序列。用户输错命令时按退格退格符留在输入流里按方向键查看历史命令时方向键序列也被当成命令内容记录。解决必须先清洗再拆分。上一章写的 CommandSplitter 就是为此准备的正则去掉 \x1b[[0-9;?]*[a-zA-Z] 控制序列删除 \x7fDEL 退格遇到行清屏序列 \r\x1b[K 整行丢弃。清洗后按 \r 或 \n 拆才能拿到真正的命令文本。5.4 并发连接数一多整个堡垒机卡死现象同时在线会话超过五个之后新用户连不上堡垒机老用户的命令也迟钝甚至断掉。原因全局共享一个 SSHClient。paramiko 的 transport 有独立的读线程和状态机多个 channel 复用一个 transport 时一个 channel 的阻塞操作会拖慢整个传输层。很多从单个会话调试起步的代码很容易把 SSHClient 写成模块级单例。解决每个用户会话独立 new 一个 SSHClient独立 transport独立 channel再放进线程池。线程池最大数以不超过机器 CPU 核心数的 2 倍为宜超过这个量进一步加线程只是增加切换开销。真到上百并发建议换 asyncio asyncssh 方案但那是另一个技术栈了。5.5 目标机断网用户卡在「正在连接」超过 5 分钟现象目标机宕了或网络被隔离用户选资产后一直卡在连接界面没有超时提示也无法主动退出。原因paramiko 的 connect 在没有指定 timeout 时会无限期等待 TCP 建连完成。尤其目标机 IP 可达但 SSH 端口不通时TCP 层会反复重试直到操作系统默认超时通常 127 秒以上才返回。解决connect 时把 timeout、banner_timeout、auth_timeout 三个参数都显式设置。TCP 建连失败走 timeout协议握手卡死走 banner_timeout认证卡死走 auth_timeout。我一般会再加一个总体连接预算比如尝试 15 秒内必须建立会话否则销毁 channel 并返回错误信息给用户。6. 进阶把一个脚本变成能一直跑的服务写到这里堡垒机核心功能都有了但一个只有脚本形态的堡垒机是没法天天挂在生产环境里的。最后一公里包含三件事进程托管、配置热加载、并发压测。6.1 进程托管systemd 与 supervisor 二选一我一般用 systemd理由是不用多装东西日志直接走 journalctl。一个最小 unit 文件长这样[Unit] DescriptionPython Bastion Service Afternetwork.target [Service] ExecStart/usr/bin/python3 /opt/bastion/main.py --config /etc/bastion/config.json WorkingDirectory/opt/bastion Restartalways RestartSec5 EnvironmentPYTHONUNBUFFERED1 [Install] WantedBymulti-user.targetRestartalways 和 RestartSec5 是核心。前者保证进程意外退出后自动拉起后者避免无间隔重启导致日志刷屏。PYTHONUNBUFFERED1 让 print 日志实时进 journald排障时不用等缓冲。如果你用的发行版默认 python 指向 3.11 或更高版本语法上注意不要用老代码里的 distutils 模块换成 setuptools 或 sysconfig。6.2 配置热加载改 JSON 不用重启服务堡垒机跑起来之后资产和角色的变更非常频繁。每次都重启服务会中断在线会话这在生产环境不可接受。热加载的做法是配置文件带 mtime 检查每次用户发起连接时读取一次配置文件的最后修改时间如果变了就 reload 到内存。import json, os, threading class ConfigStore: def __init__(self, path): self.path path self.mtime os.path.getmtime(path) self.data self.load() def load(self): with open(self.path, r) as f: return json.load(f) def refresh_if_changed(self): new_mtime os.path.getmtime(self.path) if new_mtime ! self.mtime: self.mtime new_mtime with open(self.path, r) as f: self.data json.load(f) return True return False注意 config.json 写入时要用临时文件再 rename因为 mtime 检查粒度是秒直接编辑写一半会读到残缺配置。这个 ConfigStore 加一个线程级单例配合 GIL 也不用担心多线程同时读写的可见性问题。6.3 压测脚本与上线自检部署上线前我强制自己跑一遍压测脚本用 paramiko 直接开 50 个并发会话每个会话上面循环执行命令观察 CPU、内存和网络延迟。压测的两个硬指标单会话 CPU 不超过 0.5% 核50 并发时平均命令回显延迟小于 500ms。压测脚本写起来不大但很有用import paramiko import threading import time def stress_one(idx): ssh paramiko.SSHClient() ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) ssh.connect(127.0.0.1, port2200, usernametester, passwordtest, timeout5) stdin, stdout, stderr ssh.exec_command(echo stress_%d % idx) print(stdout.read().strip()) ssh.close() threads [threading.Thread(targetstress_one, args(i,)) for i in range(50)] start time.time() for t in threads: t.start() for t in threads: t.join() print(elapsed:, time.time() - start)压测通过之后还有一个习惯我保留了很久每次改代码强制走一遍「systemctl restart bastion → 压测 → 手动登录一次做录像回放验证」三步流程。有一次我漏掉了第三步结果会话正常、录像也记录了但回放接口的索引表没更新导致线上事故复盘时找不到对应录像排查了整整半个晚上。从那以后这个三步流程成了我上线的必修课任何堡垒机相关的改动都要全套走一遍才敢收工。希望帮到你。本文还有配套的精品资源点击获取