简介：这是一份面向有一定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。这两者的差异直接决定你的堡垒机是「能跑」还是「能用」。

我一般会强制要求堡垒机场景里用 invoke_shell + get_pty。原因很朴素：运维人员会敲 vi、看 top、进 python 解释器，这些操作离开 PTY 直接退化。exec_command 只适合监控类或巡检类的非交互任务，混用会出现「用户进不了数据库客户端」「top 显示乱码」这类问题。

这里有一个常见误区：很多初版代码喜欢把 exec_command 当默认选项，因为它的返回流好读，适合做 pytest 测试。但如果你的堡垒机用户要操作真实环境，exec_command 拿到的数据流没有终端控制序列，连 top 的刷新都收不到。判断标准很简单：用户会不会在会话里按方向键、Ctrl+C、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, port=port, username=username, password=password, timeout=10, banner_timeout=15, auth_timeout=15) return ssh def open_shell(ssh): channel = ssh.invoke_shell(term='xterm') # 分配伪终端并指定窗口尺寸，防止 tab 自动补全或 top 排版错乱 channel.get_pty(term='xterm', width=160, height=48) channel.setblocking(False) return channel

这段是后面所有功能的地基。set_missing_host_key_policy(AutoAddPolicy) 用于自动接受目标机 host key，内网场景够用；如果合规要求严格，改成 WarningPolicy 并记录异常指纹更稳妥。timeout=10 是 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_timeout=300): 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: break

select.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_id=target_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 流里是 b'cd /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', errors='replace') 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[idx+1:] if line.strip(): commands.append(line.strip()) return commands

CommandSplitter 的输出会以列表形式追加进审计日志。要注意 \r 和 \n 出现的顺序对拆分结果有影响，Windows 风格的回车换行会连续触发两次，所以拆完后要 strip 再记。这个方案只对普通 Shell 命令可靠，进了 vim 或者 less 之后，控制序列的拆分会变复杂。我的习惯是：进交互子程序之后只记原始流，出子程序再恢复命令级审计。判别方式是用正则匹配 vim/less/top 这类进程启动提示符。

3.3 审计数据归档与回放接口

录像文件攒多了会面临一个问题：按什么命名、怎么快速找到某次事故的记录。我设计了一套归档规范，三张表直接对应文件系统的三个目录。

元数据记录用 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 登录到连接的完整时序

把上面两节串起来，一次完整操作请求链路是这样的：

1. 用户通过 SSH 连接堡垒机，完成登录认证（密码或公钥）。
2. 堡垒机读取用户角色和资产列表，返回可见资产菜单。
3. 用户选择一个资产，堡垒机执行授权检查：IP 是否在白名单、资产是否属于角色。
4. 建立堡垒机到目标机的 SSH 连接，分配 PTY，启动 forward。
5. forward 循环里挂上 CommandGuard，每条输入命令先过黑名单，命中则中断会话并记录告警。
6. 用户断开或超时，会话关闭，录像文件落盘，审计索引写入。

其中第 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 字节的分片大小没做细分。用户输入通常一次只有 1～3 字节，而回显往往几百字节一次。如果 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] 控制序列，删除 \x7f（DEL 退格），遇到行清屏序列 \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] Description=Python Bastion Service After=network.target [Service] ExecStart=/usr/bin/python3 /opt/bastion/main.py --config /etc/bastion/config.json WorkingDirectory=/opt/bastion Restart=always RestartSec=5 Environment=PYTHONUNBUFFERED=1 [Install] WantedBy=multi-user.target

Restart=always 和 RestartSec=5 是核心。前者保证进程意外退出后自动拉起，后者避免无间隔重启导致日志刷屏。PYTHONUNBUFFERED=1 让 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', port=2200, username='tester', password='test', timeout=5) stdin, stdout, stderr = ssh.exec_command('echo stress_%d' % idx) print(stdout.read().strip()) ssh.close() threads = [threading.Thread(target=stress_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 → 压测 → 手动登录一次做录像回放验证」三步流程。有一次我漏掉了第三步，结果会话正常、录像也记录了，但回放接口的索引表没更新，导致线上事故复盘时找不到对应录像，排查了整整半个晚上。从那以后，这个三步流程成了我上线的必修课，任何堡垒机相关的改动都要全套走一遍才敢收工。希望帮到你。

本文还有配套的精品资源，点击获取