手里管着几十台Linux服务器，最头疼的不是服务部署，而是安全基线能不能统一。今天聊的这套东西，核心是用Fabric把安全加固和合规检查攒成一个聚合器来用，统一跑、统一收、统一出报告。这篇文章就来拆解一下，这个“统一合规性聚合器”是怎么一步步设计和落地的，适合正在被安全审计折腾的运维同学，也适合想给服务器批量做安全基线加固的朋友。
我先把话说在前头：Fabric只是一个远程执行工具，它本身不懂什么是安全，真正干活的是我们写的检查项和加固动作。聚合器的价值在于把分散在几十台机器上的检查结果汇总成一张可读性很强的总表，让你知道哪台机器哪一项没过，以及为什么没过。围绕这个目标，下面的内容会从方案选型讲起，把检查项设计、Fabric脚本实现、结果聚合与报表输出，以及我实际踩过的坑全部摊开说。
1. 方案选型：为什么是Fabric来做合规聚合
1.1 现状痛点：合规检查为什么难落地
很多团队的现状是：安全基线文档写了一大堆，但落到服务器上，每台机器的状态千差万别。有人手动改过SSH配置，有人装完系统就没动过防火墙，有人系统里还跑着弱密码账户。审计一来，拿不出全量的合规报告，只能在每台机器上敲一遍命令，手动比对基线要求，效率低得让人头皮发麻。
我之前接触过的一个场景里，审计要求提供全部服务器的SSH安全配置、密码策略、文件权限等检查结果，时间只有三天。当时唯一能想到的“自动化”就是写一堆shell脚本，然后一台一台ssh进去执行，再把输出手工汇总到Excel里。这种方式的痛苦在于：脚本本身不统一，某些机器执行报错也不知道；检查项和输出格式没有标准化，统计的时候要人工反复核对；最致命的是，脚本执行完没有留痕，审计问起来说不清是哪天跑的、用的哪个版本。事实上这类工作如果靠手动，本质上是不可能持续做下去的。一两台机器还能照顾到，几十台上百台之后，合规检查就沦为了应付式抽查。
所以做这个聚合器的第一个需求其实很朴素：把检查动作、执行范围、结果收集三件事标准化，让合规状态变成随时可查询、可量化的数据，而不是存在于某台机器上的临时文件里。
1.2 Fabric的特点与选型理由
选工具的时候，我对比过Ansible、Puppet这类配置管理工具，也考虑过直接写一套基于paramiko的脚本。Ansible很强大，它自带的模块确实能省不少事，但有一个问题：引入Ansible意味着要维护一套playbook工程，对于“只想快速把合规检查和加固跑起来”的场景来说，有点杀鸡用牛刀。Puppet和Chef这种偏向持续收敛状态的方案，更适合作为长期配置管理平台，初期建设成本和运维成本都不低。直接写paramiko脚本，连接管理、命令执行、错误处理都要自己造轮子，轮子造完基础功能才刚能用，也谈不上舒服。
Fabric正好卡在中间：它本质上是一个基于SSH的远程命令执行库，用Python写脚本，比shell脚本好维护得多。Fabric 2.x的API设计得很清爽，核心就是Connection和run这两个概念。它不尝试替你管理服务器状态，只负责把你给的命令稳定地送到目标机器上执行，然后把结果带回来。这种“只做执行”的定位恰恰符合我需要的设计——合规检查的核心逻辑本来就应该由我们来定义，工具只需要保证执行和结果返回的稳定性。
另外还有一点，Fabric直接构建在paramiko之上，底层SSH的host key管理、连接复用、sftp这些都已经处理好了，不需要重复踩一遍ssh连接的那些坑。对于团队里主要是Python背景的同学来说，用Fabric写出来的代码可读性和可维护性远高于一堆shell脚本。
1.3 整体架构设计
整个聚合器的结构是这样的：一台控制机作为执行入口，上面跑Fabric脚本；若干台Linux服务器作为被管对象，控制机通过SSH连接到这些机器。控制机上还有一个结果存储层，我用的是本地SQLite，因为轻量、单文件、不需要额外部署服务。Fabric脚本执行完检查项后，会把结果写入SQLite，再由聚合模块从数据库里读出所有主机的结果，生成统计报表。
控制机 ├── fabric脚本（检查与加固） ├── sqlite_db（检查结果存储） └── report模块（聚合与输出） │ │ SSH ▼ 服务器1 服务器2 服务器3 ...这个架构的好处在于，被管端不需要安装任何额外的agent，只要控制机能ssh过去，有执行命令的权限，就能纳入检查范围。对于很多暂时不方便装agent的存量环境来说，这是最友好的接入方式。同时，因为SQLite里存的是标准化后的结构化结果，后续不管是制作HTML报告、导出CSV，还是接进监控平台的告警逻辑，都很容易扩展。
2. 合规基线与检查项设计
2.1 检查项分类体系
做合规聚合器，最核心的工作其实是定义“怎么样算合规”。这个定义必须既贴合安全基线的通用要求，又能体现自己环境里的实际控制项。我参考公开的安全基线要求，再结合自己环境里的实际控制项，把检查项分成了六类：系统基本信息、用户与认证、SSH安全、网络与防火墙、文件系统权限、内核安全参数、审计与日志。每类下面又拆出具体的检查项，每个检查项包含：唯一标识、检查命令、预期通过结果、加固命令、超时时间。
到这里可能有人会问，为什么要把“系统基本信息”也放进来。原因很简单：聚合报告里如果只有合规项的通过与否，审计老师看的时候看得很累，他会想确认这套检查到底是在什么系统版本上跑的。把内核版本、操作系统发行版这类信息作为“信息类”检查项放进去，报告的可信度会高很多。
2.2 核心检查项落地命令
直接上干货，以下是我实际用到的部分核心检查项清单，这些命令都在CentOS 7/8以及Rocky Linux上实测过。
| 检查项ID | 检查内容 | 检查命令 | 预期通过 |
|---|---|---|---|
| sys_kernel | 内核版本 | uname -r | 输出非空 |
| ssh_protocol | SSH协议版本 | grep -E '^Protocol' /etc/ssh/sshd_config | 包含Protocol 2 |
| ssh_root_login | 禁止Root直接登录 | grep -E '^PermitRootLogin' /etc/ssh/sshd_config | PermitRootLogin no |
| ssh_password_auth | 密码认证策略 | grep -E '^PasswordAuthentication' /etc/ssh/sshd_config | PasswordAuthentication no |
| pass_max_days | 密码最大有效期 | grep -E '^PASS_MAX_DAYS' /etc/login.defs | PASS_MAX_DAYS 90 |
| pass_min_len | 密码最小长度 | grep -E '^PASS_MIN_LEN' /etc/login.defs | PASS_MIN_LEN 8 |
| file_passwd_perm | passwd文件权限 | stat -c '%a' /etc/passwd | 644 |
| file_shadow_perm | shadow文件权限 | stat -c '%a' /etc/shadow | 000 或 600 |
| sysctl_ip_forward | 禁止IP转发 | sysctl -n net.ipv4.ip_forward | 0 |
| sysctl_syn_cookies | 启用SYN Cookies | sysctl -n net.ipv4.tcp_syncookies | 1 |
| auditd_status | 审计服务状态 | systemctl is-active auditd | active |
| firewalld_status | 防火墙状态 | systemctl is-active firewalld | active |
这里有个设计细节值得说一下：每条命令都是针对“单点事实”的查询，输出最好只有一行，这样比对预期值时逻辑最简单。避免在一条命令里做太复杂的组合判断，否则结果解析的时候会有很多边缘情况。如果某个配置项需要多条命令综合判断，我会拆成多个检查项，各自独立记录结果。比如SSH安全就拆了协议版本、root登录、密码认证、端口等好几个检查项。
2.3 检查项与加固动作的对应关系
检查只是第一步，发现问题之后要能改。聚合器里的每个检查项都配了一个可选的加固动作，Fabric脚本执行检查后发现不合规，可以提示用户，也可以在指定模式下自动执行加固。这里必须强调一个原则：加固动作默认不自动执行，除非用户显式传入--apply-fix参数。因为安全加固里的很多操作（比如禁用密码登录、重启sshd）是有业务影响的，挂掉一个ssh连接就麻烦了。
加固动作的写法也很讲究。比如修改/etc/login.defs里的密码策略，不能简单echo覆盖整个文件，而是用sed -i只替换关键行，并且先备份原文件到/root/bak/目录。同样的原则适用于所有涉及配置文件修改的加固项，先备份，再修改，最后重启服务。这个习惯在批量场景里能救命，尤其是当你同时加固几十台机器的时候。
3. Fabric实战：批量检查与自动加固
3.1 环境准备与连接配置
Fabric 2.x的安装很简单，一条pip命令就搞定：
pip install fabric连接配置我建议统一放在一个配置文件里维护，而不是写死在代码中。我是用YAML文件存主机列表和连接参数，格式类似下面这样：
servers: - host: 192.168.10.21 user: ops port: 22 - host: 192.168.10.22 user: ops key_filename: ~/.ssh/id_rsa在Python里用yaml.safe_load读进来，再逐个创建Connection对象。之所以用YAML而不是直接写在Python里，是为了让后续增删服务器不用改代码，只改配置文件，对不熟悉Python的同事也更友好。
创建连接时有个小点需要注意：Fabric 2.x里connect_kwargs要传入key_filename，而不是直接把密钥路径放在Connection的初始化参数里，稍微容易踩空。
from fabric import Connection def get_connection(server_info): return Connection( host=server_info["host"], user=server_info["user"], port=server_info.get("port", 22), connect_kwargs={ "key_filename": server_info.get("key_filename", "~/.ssh/id_rsa"), }, )3.2 批量采集脚本实现
检查项的采集逻辑其实很直白：遍历所有检查项，对每个检查项执行对应的命令，比对预期值，记录结果。关键要让Fabric的run能从容面对命令失败的情况，比如某条命令返回非零退出码，我们不希望整个脚本直接崩溃，而是要把失败本身记录下来。
from fabric import Connection def run_item_check(conn, item, timeout=10): """执行单个检查项，返回标准化结果""" try: result = conn.run( item["cmd"], warn=True, hide=True, timeout=timeout ) output = result.stdout.strip() status = "PASS" if item["pass"] in output.lower() else "FAIL" return { "host": conn.host, "item_id": item["id"], "item_name": item["name"], "status": status, "output": output, "expected": item["pass"], } except Exception as exc: return { "host": conn.host, "item_id": item["id"], "item_name": item["name"], "status": "ERROR", "output": str(exc), "expected": item["pass"], }warn=True表示命令返回非零时不抛异常，hide=True则抑制命令本身的输出，避免控制台被大量原始输出刷屏。timeout参数很重要，后面我会在坑点部分详细展开，当目标机器负载异常或SSH响应变慢时，没有超时限制的检查可能卡住整个流程。
遍历主机和检查项的代码长这样：
def collect_all(servers, check_items): results = [] for srv in servers: conn = get_connection(srv) for item in check_items: results.append(run_item_check(conn, item)) conn.close() return results这段代码还可以改进，比如用ThreadPoolExecutor做并行连接，控制在10个左右并发，整体检查时间能大幅缩短。并发千万不要拉得太高，否则大量同时建立SSH连接，对目标机器的sshd可能造成压力，SSH连接还会失败。
3.3 加固任务的幂等实现
加固脚本和检查脚本的差异，不只是“把配置改了就行”，更重要的是要设计成幂等：同一台机器，脚本跑两遍结果应该一致，不会因为重复执行产生副作用。也就是说，如果一次加固后目标已经是期望状态，再次执行时不应当报错，也不应当重复产生垃圾配置。
比如禁用root直接登录这个动作，我会这么写：
def fix_ssh_root_login(conn): conn.run("mkdir -p /root/bak", warn=True) conn.run("cp /etc/ssh/sshd_config /root/bak/sshd_config.bak", warn=True) conn.run( "sed -i 's/^#*PermitRootLogin.*/PermitRootLogin no/' /etc/ssh/sshd_config", warn=True, ) conn.run("systemctl restart sshd", warn=True)这里sed -i的正则同时匹配了“被注释的行”和“已经存在的配置行”，不论原来是什么状态，执行后都变成PermitRootLogin no，这就满足了幂等要求。但如果原文件里这一行被注释掉了，sed匹配后会把注释也替换掉，目标行生效，重启sshd后也不会报错。测试阶段可以先在不重要的机器上跑两遍，确认结果一致再全量执行。
配置密码策略那类加固也可以用类似方式：
def fix_password_policy(conn): conn.run("sed -i 's/^PASS_MAX_DAYS.*/PASS_MAX_DAYS 90/' /etc/login.defs", warn=True) conn.run("sed -i 's/^PASS_MIN_LEN.*/PASS_MIN_LEN 8/' /etc/login.defs", warn=True)注意如果原文件里没有PASS_MIN_LEN这一行，sed就找不到替换目标，这时候还需要追加行，所以在加固前要检查一下文件现有状态，或者用一个更稳的做法：先用grep判断再决定执行sed还是echo追加。
3.4 并行执行提升效率
几十台服务器串行执行检查，每台哪怕只有15个检查项，每项平均3秒，算下来就要半小时以上。加上网络延迟会更慢。并行是这个工具落到实际生产环境默认要考虑的事情。
我用了concurrent.futures.ThreadPoolExecutor，设置最大线程数max_workers=10。每台主机作为一个任务单元，内部仍然顺序执行该主机的所有检查项。这样既控制了并发总量，又让单台主机内的检查结果按顺序排列，后续聚合时不会搞乱脉络。
from concurrent.futures import ThreadPoolExecutor, as_completed def run_one_host(server, check_items): conn = get_connection(server) results = [] try: for item in check_items: results.append(run_item_check(conn, item)) finally: conn.close() return results def collect_parallel(servers, check_items): all_results = [] with ThreadPoolExecutor(max_workers=10) as executor: future_map = {executor.submit(run_one_host, srv, check_items): srv for srv in servers} for future in as_completed(future_map): all_results.extend(future.result()) return all_results用as_completed的好处是，哪台机器先返回就先处理哪个结果，不需要等最慢的那个跑完。出错的任务通过future.result()抛出的异常捕获后，记录到该主机的ERROR结果里，不影响其他主机的检查。这个实现看起来简单，但很实用。
4. 合规聚合与报告输出
4.1 检查结果数据模型
采集出来的数据结构是扁平的列表，每一条是一个检查项在某台机器上的结果。聚合之前，先把这些数据持久化到SQLite。表结构其实不复杂，按“一次检查任务+一个检查项+一台主机”的粒度来存：
CREATE TABLE check_results ( task_id TEXT, host TEXT, item_id TEXT, item_name TEXT, status TEXT, output TEXT, expected TEXT, check_time TIMESTAMP );加一个task_id字段，每条记录属于哪一次检查任务一目了然。这样可以保留多次检查的历史数据，后续做趋势分析时，能看出某台机器的合规状态是在变好还是变差，不会等到审计才发现一堆问题堆积到最后一刻。实际跑的时候，我会在脚本开头生成一个类似20240615_1530的task_id，带上时间和序号，保证可唯一标识。
4.2 聚合统计实现
聚合的逻辑其实不复杂，但对展示要求比较高。我需要从数据库里读出所有结果，按主机分组、按检查项分类，然后产生三个层次的视图：全局合规率、每台主机的合规率、每个检查项的通过率。
import sqlite3 def aggregate_stats(task_id): conn = sqlite3.connect("compliance.db") cur = conn.cursor() cur.execute(""" SELECT status, COUNT(*) FROM check_results WHERE task_id = ? GROUP BY status """, (task_id,)) rows = cur.fetchall() stats = {"total": 0, "pass": 0, "fail": 0, "error": 0} for status, cnt in rows: stats["total"] += cnt stats[status.lower()] = cnt conn.close() return stats更完整的聚合会按主机和检查项分组各查一次。我习惯生成三个结果集：一个总览统计、一个主机维度明细表、一个检查项维度明细表。这三个结果集分别对应报告的前段、中段和后段，审计的人看起来结构很清晰。
4.3 报告输出与展示
有了结构化的结果，输出报告就只是个模板问题。我在实现中选择了两种输出：CSV和HTML。CSV适合给别人继续做数据处理，HTML适合直接打开看。HTML报告我用一个简单的Python脚本渲染出一个自包含页面，不依赖外网CDN，公司内网离线环境也能正常打开，这个点在实际使用中相当重要。
HTML报告的核心部分是一个红色/绿色状态汇总表：每行是一台主机，每列是一个检查项，单元格显示PASS或FAIL。整块表格用颜色区分，红色醒目，绿色安心。全局合规率用一个大数字显示在报告顶部，一眼就知道整体状况。
def generate_html_report(stats, host_rows, task_id): html = [] html.append("<!DOCTYPE html><html><head><meta charset='utf-8'>") html.append("<title>Compliance Report %s</title></head><body>" % task_id) html.append("<h1>合规性检查报告 %s</h1>" % task_id) html.append("<p>全局合规率: %.1f%%</p>" % (100.0 * stats["pass"] / stats["total"])) # 渲染主机明细表 ... html.append("</body></html>") return "\n".join(html)报告生成后，脚本会把CSV和HTML都保存到reports/目录下，文件名带上task_id。这样每次检查的结果都被留存，审计来的时候，直接说出“这是某月某日全量检查的报告”，可信度完全不一样。
5. 踩坑实录与排查经验
5.1 连接与权限类问题
第一批遇到的问题是SSH连接失败。一种典型场景是控制机的known_hosts里没有目标主机的指纹，连接时会交互式询问是否信任，而Fabric脚本是非交互的，直接抛异常。解决办法是在connect_kwargs里设置disabled_algorithms或者传入known_hosts参数，更简单的做法是在初始化连接时设置connect_kwargs={"known_hosts": "~/.ssh/known_hosts"}，让paramiko自动读取。还有一种做法是主动用conn.open()测试连接，如果失败就记录到ERROR结果，不让整台主机中断。
权限问题是另一个大头。很多检查项要读/etc/shadow、要执行sysctl -w、要重启服务，普通用户跑不了。我在设计上要求控制机使用一个具有sudo权限的账号，但sudo执行也分情况：有的环境sudo需要输入密码，这就麻烦了。Fabric里可以这样处理：
conn.run("sudo cat /etc/shadow", pty=True, warn=True)如果sudo不需要密码，直接能通过；如果sudo需要密码，pyt=True会进入交互模式，但脚本里没有交互能力，所以还是要靠配置sudo免密。一个务实的做法是提前在所有被管机上配置好指定账号的sudo免密权限，这是批量运维工具的基础设施，也算不上额外负担。
5.2 检查结果不准确问题
检查项的结果不准，很多时候是命令输出格式的问题。比如grep查配置项，如果文件里同时存在“被注释的行”和“生效的行”，只有grep会把你想要的那行也匹配出来，需要结合grep -E '^PermitRootLogin'这种只匹配行首的写法，才能避开注释行。但如果原配置用的不是行首空格顶格而是前面有缩进，可能需要再多加判断。这类细节在做检查项时要多加验证，不能想当然。
另一个典型问题是sysctl -n的输出末尾会带换行，strip之后没问题，但如果某些配置项的输出有多个值或需要精确匹配，就不能简单用“包含”判断了。我在数据模型里预留了output和expected字段，就是为了让排查的人能一眼看到实际输出和预期值，定位问题时不需要重新登录机器去验证。
5.3 加固回滚与安全护栏
加固操作里最不让人放心的是“改坏了怎么办”。我在设计加固模块时，强制给每个修改配置的加固动作增加了备份步骤，备份文件统一放在/root/bak/，文件名带上主机名和时间戳。回滚脚本也写好了，按主机执行cp /root/bak/sshd_config.bak /etc/ssh/sshd_config && systemctl restart sshd就能恢复到加固前状态。
还有一个安全护栏必须强调：加固项默认不执行，只有显式传参才执行。我甚至建议在脚本入口处打印警告，要求操作者二次确认。我实盘里一次不小心把“禁用密码登录”的加固项直接跑到了批量环境里，幸好只是测试服务器，但也足够吓出一身冷汗。从那以后，我把加固执行拆成了独立脚本，只允许传入指定的目标主机列表，不能跑到所有主机上，这算是一个吃过的亏换来的经验。
另外，加固过程中如果某一台机器的sshd重启失败，要马上停止该批次中后续对同一类配置的修改，防止多台机器同时陷入不可用状态。我用一个简单的状态标记来实现：某台机器加固后反馈失败，就把它标记出来，跳过这台机器剩下的加固动作，等手动介入。
写到最后的一些体会
合规检查这种事情，做一次容易，难的是每次审计来之前都能快速给出准确的结果。我这套基于Fabric的聚合器，最实际的价值不是技术多花哨，而是把检查流程固定下来了：主机列表在YAML里维护，检查项在Python里维护，结果全部进SQLite，报告一键生成，历史记录随时可查。整个过程不再依赖某个人临时敲命令，也不怕人员变动带走脑子里那些执行细节。
有一点我很想分享给正在做同类事情的同学：不要一上来就追求把检查项做得非常多非常细，先把体系跑通比什么都重要。先做10个最关键的检查项，全流程跑通，让审计看到报告雏形，再根据实际需求慢慢扩展检查项。功能是长出来的，不是一次设计出来的。这个聚合器我用到现在，检查项已经从第一批的十几个迭代到了几十个，每一步迭代都是因为现实有需要，而不是为了显得“功能全面”。安全加固这事儿，最后拼的还是执行的可重复性和结果的可追溯性。
