企业官网建设流程全解析

简介：针对有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 user@10.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, port=BASTION_PORT, username=BASTION_USER, key_filename=BASTION_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(target=pump, args=(conn, chan), daemon=True).start() threading.Thread(target=pump, args=(chan, conn), daemon=True).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(f"listening on {LOCAL_HOST}:{LOCAL_PORT}") while True: conn, _ = listener.accept() threading.Thread(target=handle_client, args=(conn, transport), daemon=True).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 deploy@127.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(server=server) 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, username=TARGET_USER, key_filename=TARGET_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(target=bridge, args=(chan, target_ch, logfile), daemon=True).start() threading.Thread(target=bridge, args=(target_ch, chan, logfile), daemon=True).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(target=handle_connection, args=(client_sock,), daemon=True).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，作为录屏文件名，同时在审计表里记录下面的字段。

每条会话在开始时插入一条记录，结束时更新 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 user@127.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、文件路径、摘要写进审计库。回放工具打开文件时重新计算，比对不一致就标记告警。

生产环境我会再进一步：每会话写文件时边写边算哈希，会话结束后先记摘要再归档。使用上有些团队直接把摘要链成哈希链，让任何一条记录的修改都会破坏后续所有记录，取证时更有说服力。这个做法不复杂，但把审计从“能看”升级成了“可信”。

我自己最深的教训是：早期那版只做了转发，上线后被人拿着日志问“这操作是不是某人干的”，我却拿不出任何内容证据。从那以后，我把录屏和元数据落库放在了所有功能之前。堡垒机的价值不在于链路通不通，而在于出了事你能不能拿出让人信服的记录。这套方案不完美，但足够把链路、审计和回放串起来跑通，剩下的权限粒度、账号映射、审批流，都可以在这个骨架上慢慢长。希望帮到你。

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