第319篇：配置备份与合规检查自动化工具
关键词：配置备份、配置归档、合规检查、配置基线、变更检测、Netmiko、版本管理、自动化审计
一、引言
第318篇介绍了如何用 Python 自动化巡检设备运行状态。设备配置是网络运维的另一个核心资产——配置丢失可能导致网络中断，未授权变更可能引入安全隐患。因此，配置备份和合规检查是自动化运维的两大基石。
配置备份解决"丢了怎么办"的问题，合规检查解决"改得对不对"的问题。本篇将介绍如何构建一个自动化的配置备份与合规检查系统。
二、核心概念
2.1 配置备份策略
| 备份策略 | 说明 | 优点 | 缺点 |
|---|---|---|---|
| 全量备份 | 每次保存完整配置 | 简单可靠，恢复方便 | 存储占用大 |
| 增量备份 | 仅保存与上次的差异 | 存储节省 | 恢复需回溯所有增量 |
| Git 版本管理 | 每次配置提交为 Git commit | 版本追溯、差异对比、回滚 | 需 Git 仓库 |
2.2 合规检查框架
2.3 合规检查规则分类
| 规则类别 | 检查项 | 严重级别 |
|---|---|---|
| 命名规范 | 接口描述是否遵循命名模板 | Warning |
| 安全策略 | 是否启用 AAA 认证、是否禁用 Telnet | Critical |
| 协议配置 | BGP 是否启用 MD5 认证 | Critical |
| 冗余检查 | 是否有 NTP 服务器配置 | Warning |
| 资源限制 | 路由表是否超过设备容量 80% | Warning |
| 账号审计 | 是否存在默认账号/弱密码 | Critical |
三、技术细节
3.1 配置备份系统
3.1.1 基础备份脚本
# scripts/config_backup.py
from netmiko import ConnectHandler
import yaml
import os
import hashlib
from datetime import datetime
class ConfigBackup:
"""配置备份管理器"""
def __init__(self, backup_dir="backups", inventory_file="devices/inventory.yaml"):
self.backup_dir = backup_dir
os.makedirs(backup_dir, exist_ok=True)
os.makedirs(f"{backup_dir}/diffs", exist_ok=True)
with open(inventory_file, "r", encoding="utf-8") as f:
self.devices = yaml.safe_load(f).get("devices", [])
def backup_single_device(self, device):
"""备份单台设备配置"""
hostname = device.get("hostname", device["host"])
ip = device["host"]
try:
connection = ConnectHandler(**device)
# 获取完整配置
config = connection.send_command(
"display current-configuration",
read_timeout=120
)
connection.disconnect()
# 计算配置哈希
config_hash = hashlib.md5(config.encode()).hexdigest()[:12]
# 保存备份文件
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
filename = f"{hostname}_{timestamp}_{config_hash}.txt"
filepath = os.path.join(self.backup_dir, filename)
with open(filepath, "w", encoding="utf-8") as f:
f.write(config)
print(f"✓ {hostname}: 备份成功 ({len(config)} bytes, hash={config_hash})")
return {
"hostname": hostname,
"ip": ip,
"filepath": filepath,
"hash": config_hash,
"size": len(config),
"timestamp": datetime.now().isoformat(),
}
except Exception as e:
print(f"✗ {hostname}: 备份失败 - {e}")
return {
"hostname": hostname,
"ip": ip,
"status": "error",
"error": str(e),
}
def backup_all_devices(self, max_workers=5):
"""备份所有设备"""
from concurrent.futures import ThreadPoolExecutor, as_completed
results = []
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = {
executor.submit(self.backup_single_device, dev): dev
for dev in self.devices
}
for future in as_completed(futures):
result = future.result()
results.append(result)
return results
def get_latest_backup(self, hostname):
"""获取指定设备的最新备份"""
files = sorted(
[f for f in os.listdir(self.backup_dir) if f.startswith(hostname + "_")],
reverse=True
)
if files:
filepath = os.path.join(self.backup_dir, files[0])
with open(filepath, "r", encoding="utf-8") as f:
return f.read()
return None
3.1.2 Git 版本管理备份
# scripts/config_git_backup.py
import subprocess
import os
from datetime import datetime
class GitConfigBackup:
"""基于 Git 的配置版本管理"""
def __init__(self, repo_dir="config_repo"):
self.repo_dir = repo_dir
self._init_repo()
def _init_repo(self):
"""初始化 Git 仓库"""
if not os.path.exists(self.repo_dir):
os.makedirs(self.repo_dir)
os.makedirs(f"{self.repo_dir}/running", exist_ok=True)
os.makedirs(f"{self.repo_dir}/startup", exist_ok=True)
try:
subprocess.run(["git", "-C", self.repo_dir, "init"],
capture_output=True, check=True)
subprocess.run(["git", "-C", self.repo_dir, "config",
"user.email", "automation@example.com"],
capture_output=True, check=True)
subprocess.run(["git", "-C", self.repo_dir, "config",
"user.name", "Config Backup Bot"],
capture_output=True, check=True)
except subprocess.CalledProcessError as e:
print(f"Git 初始化失败: {e}")
def save_config(self, hostname, config, config_type="running"):
"""保存配置并提交到 Git"""
filepath = os.path.join(self.repo_dir, config_type,
f"{hostname}.txt")
# 写入配置
with open(filepath, "w", encoding="utf-8") as f:
f.write(config)
# Git 操作
subprocess.run(["git", "-C", self.repo_dir, "add", filepath],
capture_output=True, check=True)
# 提交
commit_msg = f"Backup {hostname} {config_type}: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
subprocess.run(["git", "-C", self.repo_dir, "commit", "-m", commit_msg],
capture_output=True, check=True)
print(f"✓ {hostname} {config_type} 已提交到 Git")
def get_diff(self, hostname, config_type="running"):
"""获取与上次备份的配置差异"""
filepath = os.path.join(self.repo_dir, config_type, f"{hostname}.txt")
try:
result = subprocess.run(
["git", "-C", self.repo_dir, "diff", "HEAD~1", "HEAD",
"--", filepath],
capture_output=True, text=True, check=True
)
return result.stdout
except subprocess.CalledProcessError:
return None
def get_history(self, hostname, config_type="running", max_count=10):
"""获取配置变更历史"""
filepath = os.path.join(self.repo_dir, config_type, f"{hostname}.txt")
try:
result = subprocess.run(
["git", "-C", self.repo_dir, "log", "--oneline",
f"-{max_count}", "--", filepath],
capture_output=True, text=True, check=True
)
return result.stdout
except subprocess.CalledProcessError:
return None
3.2 合规检查引擎
3.2.1 规则定义
# scripts/compliance_rules.py
import re
class ComplianceRule:
"""合规检查规则基类"""
def __init__(self, rule_id, name, severity, description):
self.rule_id = rule_id
self.name = name
self.severity = severity # critical / warning / info
self.description = description
def check(self, config):
"""执行检查，返回 (bool, str)"""
raise NotImplementedError
class SecurityRule(ComplianceRule):
"""安全类规则"""
pass
class NamingRule(ComplianceRule):
"""命名规范规则"""
pass
class ProtocolRule(ComplianceRule):
"""协议配置规则"""
pass
# ========== 具体规则实现 ==========
class CheckAAAEnabled(SecurityRule):
"""检查 AAA 认证是否启用"""
def __init__(self):
super().__init__(
rule_id="SEC-001",
name="AAA认证启用",
severity="critical",
description="设备应启用 AAA 认证"
)
def check(self, config):
if re.search(r'^aaa', config, re.MULTILINE):
return True, "AAA 认证已启用"
return False, "AAA 认证未启用"
class CheckTelnetDisabled(SecurityRule):
"""检查 Telnet 是否禁用"""
def __init__(self):
super().__init__(
rule_id="SEC-002",
name="Telnet禁用",
severity="critical",
description="设备应禁用 Telnet，仅使用 SSH"
)
def check(self, config):
if re.search(r'^undo\s+user-interface\s+type\s+telnet', config, re.MULTILINE):
return True, "Telnet 已禁用"
# 如果没有 telnet 相关配置，也算通过
if not re.search(r'user-interface\s+type\s+telnet', config, re.MULTILINE):
return True, "未发现 Telnet 配置"
return False, "Telnet 服务可能未禁用"
class CheckBGPMD5(SecurityRule):
"""检查 BGP 是否启用 MD5 认证"""
def __init__(self):
super().__init__(
rule_id="SEC-003",
name="BGP MD5认证",
severity="critical",
description="BGP 对等体应配置 MD5 认证"
)
def check(self, config):
# 查找 BGP peer 配置
peer_pattern = re.compile(r'^\s*peer\s+(\S+)', re.MULTILINE)
md5_pattern = re.compile(r'^\s*peer\s+(\S+)\s+simple-ciphertext', re.MULTILINE)
peers = peer_pattern.findall(config)
md5_peers = md5_pattern.findall(config)
if not peers:
return True, "未发现 BGP 配置"
unprotected = set(peers) - set(md5_peers)
if not unprotected:
return True, f"所有 {len(peers)} 个 BGP 对等体均配置了 MD5 认证"
return False, f"{len(unprotected)} 个 BGP 对等体未配置 MD5: {', '.join(list(unprotected)[:5])}"
class CheckInterfaceDescription(NamingRule):
"""检查接口描述是否符合命名规范"""
def __init__(self, pattern=r"^(To-|Uplink|Access|Link-)[A-Za-z0-9_-]+"):
super().__init__(
rule_id="NAM-001",
name="接口描述命名规范",
severity="warning",
description=f"接口描述应匹配模式: {pattern}"
)
self.pattern = re.compile(pattern)
def check(self, config):
# 查找接口块中的描述
interface_blocks = re.findall(
r'(interface\s+\S+\n(?:[^\n]*\n)*?)(?=interface\s+\S+|\Z)',
config, re.MULTILINE
)
non_compliant = []
for block in interface_blocks:
iface_match = re.match(r'interface\s+(\S+)', block)
desc_match = re.search(r'description\s+(.+)', block)
if iface_match and desc_match:
iface = iface_match.group(1)
desc = desc_match.group(1).strip().strip('"')
# 跳过 Loopback 和管理接口
if iface.lower().startswith(("loop", "me", "inloop")):
continue
if not self.pattern.match(desc):
non_compliant.append(f"{iface}: \"{desc}\"")
if not non_compliant:
return True, "所有接口描述符合命名规范"
return False, f"{len(non_compliant)} 个接口描述不规范: {', '.join(non_compliant[:5])}"
class CheckNTPConfigured(NamingRule):
"""检查 NTP 时间同步配置"""
def __init__(self):
super().__init__(
rule_id="CFG-001",
name="NTP配置",
severity="warning",
description="设备应配置 NTP 服务器"
)
def check(self, config):
ntp_servers = re.findall(r'^\s*ntp-server\s+(\S+)', config, re.MULTILINE)
if ntp_servers:
return True, f"NTP 已配置 {len(ntp_servers)} 台服务器"
return False, "未配置 NTP 服务器"
class CheckSNMPCommunity(SecurityRule):
"""检查 SNMP Community 字符串"""
def __init__(self):
super().__init__(
rule_id="SEC-004",
name="SNMP Community安全",
severity="critical",
description="SNMP Community 不应使用默认值"
)
def check(self, config):
weak_communities = ["public", "private", "admin", "test", "huawei", "default"]
snmp_matches = re.findall(
r'snmp-agent\s+community\s+(?:read|write)\s+(\S+)',
config, re.MULTILINE
)
weak_found = [c for c in snmp_matches if c.lower() in weak_communities]
if weak_found:
return False, f"发现弱 SNMP Community: {', '.join(weak_found)}"
if snmp_matches:
return True, "SNMP Community 已配置且非弱密码"
return True, "未配置 SNMP Community"
3.2.2 合规检查引擎
# scripts/compliance_checker.py
from compliance_rules import *
class ComplianceChecker:
"""合规检查引擎"""
def __init__(self):
# 注册所有规则
self.rules = [
CheckAAAEnabled(),
CheckTelnetDisabled(),
CheckBGPMD5(),
CheckInterfaceDescription(),
CheckNTPConfigured(),
CheckSNMPCommunity(),
]
def check_config(self, config):
"""检查配置合规性"""
results = []
for rule in self.rules:
passed, message = rule.check(config)
results.append({
"rule_id": rule.rule_id,
"rule_name": rule.name,
"severity": rule.severity,
"passed": passed,
"message": message,
"description": rule.description,
})
return results
def check_device(self, device, config):
"""检查单台设备"""
results = self.check_config(config)
violation_count = sum(1 for r in results if not r["passed"])
critical_count = sum(
1 for r in results
if not r["passed"] and r["severity"] == "critical"
)
return {
"hostname": device.get("hostname", device["host"]),
"ip": device["host"],
"timestamp": datetime.now().isoformat(),
"total_rules": len(results),
"violations": violation_count,
"critical_violations": critical_count,
"compliance_rate": round(
(len(results) - violation_count) / len(results) * 100, 1
) if results else 0,
"details": results,
}
3.2.3 批量合规检查
# scripts/batch_compliance.py
from compliance_checker import ComplianceChecker
from netmiko import ConnectHandler
import yaml
import json
def run_compliance_audit(inventory_file="devices/inventory.yaml"):
"""批量合规审计"""
with open(inventory_file, "r", encoding="utf-8") as f:
inventory = yaml.safe_load(f)
checker = ComplianceChecker()
results = []
for device in inventory.get("devices", []):
hostname = device.get("hostname", device["host"])
print(f"检查 {hostname}...")
try:
connection = ConnectHandler(**device)
config = connection.send_command(
"display current-configuration",
read_timeout=120
)
connection.disconnect()
result = checker.check_device(device, config)
results.append(result)
status = "✓" if result["violations"] == 0 else "⚠"
print(f" {status} 合规率 {result['compliance_rate']}%, "
f"{result['violations']} 违规, "
f"{result['critical_violations']} 严重")
except Exception as e:
results.append({
"hostname": hostname,
"ip": device["host"],
"status": "error",
"error": str(e),
})
return results
# 执行
audit_results = run_compliance_audit()
# 保存审计结果
with open("compliance_audit.json", "w", encoding="utf-8") as f:
json.dump({
"timestamp": datetime.now().isoformat(),
"summary": {
"total_devices": len(audit_results),
"devices_with_violations": sum(
1 for r in audit_results if r.get("violations", 0) > 0
),
"total_violations": sum(r.get("violations", 0) for r in audit_results),
},
"results": audit_results,
}, f, indent=2, ensure_ascii=False)
3.3 配置变更检测
# scripts/config_change_detection.py
import difflib
import hashlib
class ConfigChangeDetector:
"""配置变更检测器"""
def __init__(self, backup_dir="backups"):
self.backup_dir = backup_dir
def compute_hash(self, config):
"""计算配置哈希"""
# 去除时间戳等动态内容后计算
normalized = re.sub(
r'(uptime|up\s+time)[:\s]*[\d\s:,]+',
r'\1: <normalized>',
config, flags=re.IGNORECASE
)
return hashlib.md5(normalized.encode()).hexdigest()
def diff_configs(self, old_config, new_config):
"""计算配置差异"""
old_lines = old_config.splitlines(keepends=True)
new_lines = new_config.splitlines(keepends=True)
diff = difflib.unified_diff(
old_lines, new_lines,
fromfile="old_config", tofile="new_config",
n=3 # 3 行上下文
)
return "".join(diff)
def detect_changes(self, hostname):
"""检测指定设备的配置变更"""
backups = sorted(
[f for f in os.listdir(self.backup_dir)
if f.startswith(hostname + "_")],
reverse=True
)
if len(backups) < 2:
return None
# 读取最近两次备份
latest = os.path.join(self.backup_dir, backups[0])
previous = os.path.join(self.backup_dir, backups[1])
with open(latest, "r", encoding="utf-8") as f:
new_config = f.read()
with open(previous, "r", encoding="utf-8") as f:
old_config = f.read()
new_hash = self.compute_hash(new_config)
old_hash = self.compute_hash(old_config)
if new_hash == old_hash:
return {
"hostname": hostname,
"changed": False,
"message": "配置无变更",
}
diff = self.diff_configs(old_config, new_config)
return {
"hostname": hostname,
"changed": True,
"old_hash": old_hash,
"new_hash": new_hash,
"diff": diff,
"added_lines": len([l for l in diff.split("\n") if l.startswith("+")]) - 1,
"removed_lines": len([l for l in diff.split("\n") if l.startswith("-")]) - 1,
}
四、实战配置/案例分析
4.1 案例一：完整配置备份与审计流程
# scripts/backup_and_audit_pipeline.py
from config_backup import ConfigBackup
from compliance_checker import ComplianceChecker
from config_change_detection import ConfigChangeDetector
from inspection_report import generate_html_report
def run_backup_and_audit_pipeline():
"""配置备份 + 合规审计一体化流程"""
print("=" * 60)
print("配置备份与合规审计流程")
print("=" * 60)
# Step 1: 配置备份
print("\nStep 1: 配置备份...")
backup = ConfigBackup()
backup_results = backup.backup_all_devices()
success_backups = [r for r in backup_results if "filepath" in r]
print(f" 备份成功: {len(success_backups)}/{len(backup_results)}")
# Step 2: 变更检测
print("\nStep 2: 变更检测...")
detector = ConfigChangeDetector()
changes = []
for result in success_backups:
hostname = result["hostname"]
change = detector.detect_changes(hostname)
if change and change.get("changed"):
changes.append(change)
print(f" ⚠ {hostname}: 配置已变更 "
f"(+{change['added_lines']} / -{change['removed_lines']})")
# Step 3: 合规审计
print("\nStep 3: 合规审计...")
checker = ComplianceChecker()
audit_results = run_compliance_audit()
# Step 4: 汇总报告
print("\nStep 4: 生成报告...")
summary = {
"total_devices": len(backup_results),
"backup_success": len(success_backups),
"config_changes": len(changes),
"compliance_violations": sum(
r.get("violations", 0) for r in audit_results
),
}
print(f"\n{'='*60}")
print(f"汇总: {json.dumps(summary, indent=2, ensure_ascii=False)}")
return summary
run_backup_and_audit_pipeline()
4.2 案例二：故障案例——配置回滚失败
故障现象：工程师手动修改设备配置后导致路由震荡，尝试从备份恢复但发现备份文件过期（已是 3 天前的配置）。
根因分析：
- 备份频率太低（每周一次），无法反映最新配置
- 没有自动化触发机制，依赖人工执行
- 备份文件命名不规范，难以快速定位
解决方案：
# scripts/enhanced_backup.py
import schedule
from config_backup import ConfigBackup
def automated_backup_schedule():
"""增强版自动备份调度"""
backup = ConfigBackup()
# 每日凌晨 2:00 全量备份
schedule.every().day.at("02:00").do(backup.backup_all_devices)
# 变更检测（每小时）
schedule.every().hour.do(detect_and_alert_changes)
print("自动备份调度已启动")
while True:
schedule.run_pending()
time.sleep(60)
def detect_and_alert_changes():
"""检测配置变更并告警"""
detector = ConfigChangeDetector()
for device in get_device_list():
change = detector.detect_changes(device["hostname"])
if change and change.get("changed"):
# 立即触发一次备份
ConfigBackup().backup_single_device(device)
# 发送变更通知
send_change_alert(device["hostname"], change)
改进措施：
| 改进项 | 原方案 | 改进方案 |
|---|---|---|
| 备份频率 | 每周手动 | 每日自动 + 变更触发即时备份 |
| 变更检测 | 无 | 每小时自动检测，变更即时告警 |
| 备份存储 | 本地单文件 | Git 版本管理 + 远程备份 |
| 恢复时间 | 手动查找旧文件 | 一键恢复到指定版本 |
| 通知机制 | 无 | 配置变更即时邮件/企微通知 |
4.3 案例三：合规基线管理与自动修复
# scripts/compliance_baseline.py
import json
class ComplianceBaseline:
"""合规基线管理器"""
def __init__(self, baseline_file="baselines/default_baseline.json"):
self.baseline_file = baseline_file
self.baseline = self._load_baseline()
def _load_baseline(self):
"""加载基线配置"""
default_baseline = {
"rules": {
"SEC-001": {"enabled": True, "severity": "critical"},
"SEC-002": {"enabled": True, "severity": "critical"},
"SEC-003": {"enabled": True, "severity": "critical"},
"SEC-004": {"enabled": True, "severity": "critical"},
"NAM-001": {"enabled": True, "severity": "warning"},
"CFG-001": {"enabled": True, "severity": "warning"},
},
"auto_remediation": {
"NAM-001": {
"action": "fix_interface_description",
"template": "To-{peer_device}-{peer_interface}",
},
},
}
return default_baseline
def generate_remediation_commands(self, violations, device_type="huawei"):
"""根据违规项生成修复命令"""
commands = []
for violation in violations:
rule_id = violation["rule_id"]
if rule_id == "NAM-001" and device_type == "huawei":
# 自动修复接口描述
non_compliant = violation.get("non_compliant_interfaces", [])
for iface in non_compliant:
commands.append({
"interface": iface["name"],
"commands": [
f"interface {iface['name']}",
f"description {iface['suggested_description']}",
]
})
elif rule_id == "SEC-004": # SNMP weak community
commands.append({
"commands": [
"snmp-agent community read cipher <new-strong-community>",
]
})
return commands
五、总结
| 功能 | 实现方式 | 频率建议 |
|---|---|---|
| 配置备份 | Netmiko + 文件系统/Git | 每日 + 变更即时 |
| 合规检查 | 规则引擎 + 正则匹配 | 每周全量，每日增量 |
| 变更检测 | 配置哈希 + diff | 每小时检测 |
| 自动修复 | 模板化命令生成 | 按需手动确认 |
| 最佳实践 | 说明 |
|---|---|
| Git 版本管理 | 使用 Git 管理配置，支持版本追溯和差异对比 |
| 分级告警 | Critical 即时通知，Warning 汇总日报 |
| 变更关联 | 配置变更与工单关联，确保可追溯 |
| 自动化修复 | 对低风险的违规项实现自动修复，高风险需人工确认 |
| 多备份存储 | 本地 + 远程双备份，防止单点故障 |
| 基线更新 | 定期更新合规基线，适应网络架构变化 |
思考题
- 使用 Git 管理配置版本相比传统的文件备份有什么优势？有什么需要注意的地方？
- 合规检查规则应该由谁来定义？如何保证规则的及时更新？
- 自动修复配置变更的功能存在哪些风险？如何控制自动修复的范围？
下篇预告：第320篇：IP 地址管理（IPAM）自动化