第323篇：网络配置合规性检查自动化

关键词

合规检查、配置基线、安全合规、自动审计、策略验证、配置漂移检测、合规报告

一、为何要自动化合规检查

1.1 合规检查的痛点

传统合规检查的困境：

  检查频率低：
  ┌─ 大部分企业每月/每季度检查一次
  ├─ 两次检查之间配置可能已漂移
  └─ 发现问题时已不符合合规要求数周

  检查成本高：
  ┌─ 工程师手动登录每台设备
  ├─ 逐条命令核对配置
  ├─ 100 台设备 × 50 条规则 = 5000 次检查
  └─ 耗时：2-3 人天/次

  标准不一致：
  ┌─ 不同工程师对同一规则理解不同
  ├─ Excel 记录的检查结果易出错
  └─ 难以跟踪整改状态

  事后诸葛亮：
  ┌─ 等审计发现问题才整改
  ├─ 安全事件发生后才知道不合规
  └─ 合规是被动的"灭火"，不是主动预防

  自动化合规检查的价值：
  ┌─ 实时：每天甚至每小时自动检查
  ├─ 一致：同样的规则，同样的标准
  ├─ 全面：所有设备、所有规则一次覆盖
  └─ 主动：配置漂移立即告警

1.2 合规检查框架

自动化合规检查框架：

二、合规规则定义

2.1 规则格式设计

#!/usr/bin/env python3
# compliance_rules.py — 合规规则定义

from dataclasses import dataclass, field
from typing import List, Dict, Any, Callable, Optional
from enum import Enum


class Severity(Enum):
    """规则严重级别"""
    CRITICAL = "CRITICAL"   # 致命：必须立即整改
    HIGH     = "HIGH"       # 高：需尽快整改
    MEDIUM   = "MEDIUM"     # 中：建议整改
    LOW      = "LOW"        # 低：信息提示


class RuleCategory(Enum):
    """规则分类"""
    SECURITY     = "安全基线"
    COMPLIANCE   = "合规要求"
    BEST_PRACTICE = "最佳实践"
    STANDARD     = "企业规范"
    PERFORMANCE  = "性能优化"


@dataclass
class ComplianceRule:
    """合规规则定义"""
    id: str                            # 规则 ID（如 SEC-001）
    name: str                          # 规则名称
    category: RuleCategory             # 规则分类
    severity: Severity                 # 严重级别
    description: str                   # 规则描述
    device_types: List[str]            # 适用设备类型
    check_func: Callable               # 检查函数
    remediation: str                   # 整改建议
    reference: str = ""                # 参考标准

    def to_dict(self) -> Dict:
        return {
            "id": self.id,
            "name": self.name,
            "category": self.category.value,
            "severity": self.severity.value,
            "description": self.description,
            "remediation": self.remediation,
        }


@dataclass
class CheckResult:
    """检查结果"""
    rule_id: str
    device: str
    passed: bool
    actual_value: Any
    expected_value: Any
    detail: str = ""


class ComplianceResult:
    """设备合规检查结果"""
    def __init__(self, device: str, timestamp: str):
        self.device = device
        self.timestamp = timestamp
        self.results: List[CheckResult] = []
        self.summary = {"pass": 0, "fail": 0, "error": 0}

    def add_result(self, result: CheckResult):
        self.results.append(result)
        if result.passed:
            self.summary["pass"] += 1
        else:
            self.summary["fail"] += 1

    def pass_rate(self) -> float:
        total = self.summary["pass"] + self.summary["fail"]
        return self.summary["pass"] / total if total > 0 else 1.0

2.2 常用合规规则示例

#!/usr/bin/env python3
# rules_security.py — 安全基线规则

from compliance_rules import (
    ComplianceRule, RuleCategory, Severity, CheckResult,
)
import re
from typing import Dict, Any


# =============================================
# 规则定义函数
# =============================================

def check_ssh_version(config: Dict) -> CheckResult:
    """检查 SSH 版本（必须是 SSHv2）"""
    ssh_config = config.get("ssh", {})
    version = ssh_config.get("version", "unknown")
    passed = "2" in str(version)
    return CheckResult(
        rule_id="SEC-001",
        device=config.get("hostname", ""),
        passed=passed,
        actual_value=version,
        expected_value="SSHv2",
        detail=f"SSH 版本: {version}（{'合规' if passed else '不合规'}）",
    )


def check_password_policy(config: Dict) -> CheckResult:
    """检查密码策略"""
    aaa = config.get("aaa", {})
    # 检查密码加密
    password_encrypted = aaa.get("password_encryption", False)
    # 检查密码最小长度
    min_length = aaa.get("password_min_length", 0)
    passed = password_encrypted and min_length >= 8
    return CheckResult(
        rule_id="SEC-002",
        device=config.get("hostname", ""),
        passed=passed,
        actual_value=f"加密={password_encrypted}, 最小长度={min_length}",
        expected_value="加密启用, 最小长度 ≥ 8",
        detail="密码策略检查",
    )


def check_telnet_disabled(config: Dict) -> CheckResult:
    """检查 Telnet 是否已禁用"""
    services = config.get("services", {})
    telnet = services.get("telnet", True)  # 默认 true
    passed = not telnet
    return CheckResult(
        rule_id="SEC-003",
        device=config.get("hostname", ""),
        passed=passed,
        actual_value=f"Telnet={'启用' if telnet else '禁用'}",
        expected_value="Telnet 禁用",
    )


def check_ntp_config(config: Dict) -> CheckResult:
    """检查 NTP 配置"""
    ntp = config.get("ntp", {})
    servers = ntp.get("servers", [])
    passed = len(servers) >= 2  # 至少 2 个 NTP 服务器
    return CheckResult(
        rule_id="BPC-001",
        device=config.get("hostname", ""),
        passed=passed,
        actual_value=f"NTP 服务器数: {len(servers)}",
        expected_value="≥ 2 个 NTP 服务器",
    )


def check_snmp_version(config: Dict) -> CheckResult:
    """检查 SNMP 版本（禁止使用 v1/v2c）"""
    snmp = config.get("snmp", {})
    version = snmp.get("version", "v2c")
    passed = version in ("v3",)
    return CheckResult(
        rule_id="SEC-004",
        device=config.get("hostname", ""),
        passed=passed,
        actual_value=f"SNMP {version}",
        expected_value="SNMP v3",
    )


def check_logging_config(config: Dict) -> CheckResult:
    """检查日志配置"""
    logging = config.get("logging", {})
    has_server = len(logging.get("servers", [])) > 0
    has_buffer = logging.get("buffer", False)
    passed = has_server and has_buffer
    return CheckResult(
        rule_id="BPC-002",
        device=config.get("hostname", ""),
        passed=passed,
        actual_value=f"日志服务器={'有' if has_server else '无'}, "
                     f"缓存日志={'有' if has_buffer else '无'}",
        expected_value="需配置日志服务器和缓存日志",
    )


# =============================================
# 规则注册表
# =============================================

# 安全基线规则
SECURITY_RULES = [
    ComplianceRule(
        id="SEC-001",
        name="SSH 版本检查",
        category=RuleCategory.SECURITY,
        severity=Severity.CRITICAL,
        description="必须使用 SSHv2，禁止使用 SSHv1",
        device_types=["switch", "router", "firewall"],
        check_func=check_ssh_version,
        remediation="配置 ssh server version 2",
        reference="PCI-DSS 2.2",
    ),
    ComplianceRule(
        id="SEC-002",
        name="密码策略检查",
        category=RuleCategory.SECURITY,
        severity=Severity.HIGH,
        description="密码加密必须启用，最小密码长度 ≥ 8",
        device_types=["switch", "router", "firewall"],
        check_func=check_password_policy,
        remediation="配置 password encryption 和密码最小长度",
        reference="ISO 27001",
    ),
    ComplianceRule(
        id="SEC-003",
        name="Telnet 禁用检查",
        category=RuleCategory.SECURITY,
        severity=Severity.CRITICAL,
        description="Telnet 必须禁用，仅允许 SSH 登录",
        device_types=["switch", "router"],
        check_func=check_telnet_disabled,
        remediation="关闭 telnet server，仅保留 ssh server",
    ),
    ComplianceRule(
        id="SEC-004",
        name="SNMP 版本检查",
        category=RuleCategory.SECURITY,
        severity=Severity.HIGH,
        description="仅允许 SNMP v3",
        device_types=["switch", "router"],
        check_func=check_snmp_version,
        remediation="升级 SNMP 配置到 v3",
        reference="等保 2.0 三级",
    ),
]

# 最佳实践规则
BEST_PRACTICE_RULES = [
    ComplianceRule(
        id="BPC-001",
        name="NTP 配置检查",
        category=RuleCategory.BEST_PRACTICE,
        severity=Severity.MEDIUM,
        description="至少配置 2 个 NTP 服务器",
        device_types=["switch", "router"],
        check_func=check_ntp_config,
        remediation="配置至少 2 个 NTP 服务器",
    ),
    ComplianceRule(
        id="BPC-002",
        name="日志配置检查",
        category=RuleCategory.BEST_PRACTICE,
        severity=Severity.MEDIUM,
        description="需配置日志服务器和日志缓存",
        device_types=["switch", "router"],
        check_func=check_logging_config,
        remediation="配置日志服务器地址和日志缓存",
    ),
]

# 合并所有规则
ALL_RULES = SECURITY_RULES + BEST_PRACTICE_RULES

三、自动化检查引擎

3.1 采集与解析

#!/usr/bin/env python3
# compliance_engine.py — 合规检查引擎

from netmiko import ConnectHandler
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Dict, List, Any
import json
import time
import logging

logger = logging.getLogger(__name__)


class DeviceConfigCollector:
    """设备配置采集器"""

    COLLECT_COMMANDS = {
        "huawei_vrp": [
            "display current-configuration",
            "display ssh server status",
            "display ntp status",
            "display snmp-agent sys-info version",
            "display logbuffer",
        ],
        "cisco_ios": [
            "show running-config",
            "show ip ssh",
            "show ntp associations",
            "show snmp community",
            "show logging",
        ],
    }

    def collect(self, device_info: Dict) -> Dict:
        """采集单台设备配置"""
        try:
            conn = ConnectHandler(**device_info)
            conn.enable()
            hostname = conn.find_prompt().rstrip("#>")

            config_data = {
                "hostname": hostname,
                "device_type": device_info["device_type"],
                "collected_at": time.strftime("%Y-%m-%d %H:%M:%S"),
                "raw_outputs": {},
            }

            # 执行采集命令
            commands = self.COLLECT_COMMANDS.get(
                device_info["device_type"], []
            )
            for cmd in commands:
                output = conn.send_command(
                    cmd, read_timeout=60
                )
                config_data["raw_outputs"][cmd] = output

            conn.disconnect()
            return config_data

        except Exception as e:
            logger.error(
                f"采集 {device_info.get('host')} 失败: {e}"
            )
            return {"hostname": "unknown", "error": str(e)}

    def collect_all(
        self, devices: List[Dict], max_workers: int = 10
    ) -> List[Dict]:
        """并发采集所有设备"""
        results = []
        with ThreadPoolExecutor(max_workers=max_workers) as executor:
            futures = {
                executor.submit(self.collect, dev): dev
                for dev in devices
            }
            for future in as_completed(futures):
                results.append(future.result())
        return results

3.2 配置解析与结构化

class ConfigParser:
    """配置解析器：原始文本 → 结构化数据"""

    @staticmethod
    def parse_running_config(raw_output: str) -> Dict:
        """解析 running-config 为结构化数据"""
        config = {}
        current_section = "global"
        config[current_section] = {}

        for line in raw_output.splitlines():
            line = line.strip()
            if not line or line.startswith("#"):
                continue

            # 检测段落开始
            if line.startswith("interface "):
                current_section = line
                config[current_section] = {"commands": []}
            elif line.startswith("bgp "):
                current_section = line
                config[current_section] = {"commands": []}
            elif line.startswith("ospf "):
                current_section = line
                config[current_section] = {"commands": []}
            elif current_section in config:
                config[current_section]["commands"].append(line)

        return config

    @staticmethod
    def extract_structured(config: Dict) -> Dict:
        """提取关键配置项为结构化数据"""
        raw = config.get("raw_outputs", {})
        running = raw.get(
            "display current-configuration",
            raw.get("show running-config", ""),
        )
        parsed = ConfigParser.parse_running_config(running)

        structured = {
            "hostname": config.get("hostname", ""),
            "ssh": ConfigParser._parse_ssh(raw),
            "ntp": ConfigParser._parse_ntp(raw),
            "snmp": ConfigParser._parse_snmp(raw),
            "aaa": ConfigParser._parse_aaa(parsed),
            "services": ConfigParser._parse_services(parsed),
            "logging": ConfigParser._parse_logging(parsed),
            "interfaces": ConfigParser._parse_interfaces(parsed),
        }
        return structured

    @staticmethod
    def _parse_ssh(raw: Dict) -> Dict:
        ssh_output = raw.get(
            "display ssh server status",
            raw.get("show ip ssh", ""),
        )
        return {
            "version": "v2" if "v2" in ssh_output else "v1",
            "enabled": "enable" in ssh_output.lower(),
        }

    @staticmethod
    def _parse_ntp(raw: Dict) -> Dict:
        ntp_output = raw.get(
            "display ntp status",
            raw.get("show ntp associations", ""),
        )
        servers = [
            line.split()[-1] for line in ntp_output.splitlines()
            if "server" in line.lower()
        ]
        return {
            "configured": len(servers) > 0,
            "servers": servers,
        }

    @staticmethod
    def _parse_snmp(raw: Dict) -> Dict:
        snmp_output = raw.get(
            "display snmp-agent sys-info version",
            raw.get("show snmp community", ""),
        )
        if "v3" in snmp_output.lower():
            version = "v3"
        elif "v2c" in snmp_output.lower():
            version = "v2c"
        else:
            version = "v1"
        return {"version": version}

    @staticmethod
    def _parse_aaa(parsed: Dict) -> Dict:
        aaa = parsed.get("global", {})
        return {
            "password_encryption": any(
                "password encryption" in c.lower()
                for c in aaa.get("commands", [])
            ),
            "password_min_length": 8,  # 简化处理
        }

    @staticmethod
    def _parse_services(parsed: Dict) -> Dict:
        global_cmds = parsed.get("global", {}).get("commands", [])
        return {
            "telnet": any(
                "telnet server enable" in c.lower()
                for c in global_cmds
            ),
        }

    @staticmethod
    def _parse_logging(raw: Dict) -> Dict:
        log_output = raw.get("display logbuffer", "")
        return {
            "buffer": bool(log_output.strip()),
            "servers": [],  # 简化处理
        }

    @staticmethod
    def _parse_interfaces(parsed: Dict) -> List[Dict]:
        interfaces = []
        for section, content in parsed.items():
            if section.startswith("interface "):
                name = section.split(None, 1)[1]
                interfaces.append({
                    "name": name,
                    "commands": content.get("commands", []),
                })
        return interfaces

3.3 检查执行

class ComplianceChecker:
    """合规检查执行器"""

    def __init__(self, rules: List[ComplianceRule]):
        self.rules = rules

    def check_device(
        self, device_config: Dict
    ) -> ComplianceResult:
        """对单台设备执行所有合规检查"""
        result = ComplianceResult(
            device=device_config.get("hostname", "unknown"),
            timestamp=time.strftime("%Y-%m-%d %H:%M:%S"),
        )

        # 提取结构化配置
        structured = ConfigParser.extract_structured(
            device_config
        )

        for rule in self.rules:
            try:
                check_result = rule.check_func(structured)
                result.add_result(check_result)
            except Exception as e:
                logger.error(
                    f"规则 {rule.id} 检查失败: {e}"
                )
                result.add_result(CheckResult(
                    rule_id=rule.id,
                    device=device_config.get("hostname", ""),
                    passed=False,
                    actual_value=str(e),
                    expected_value="",
                    detail=f"检查异常: {e}",
                ))

        return result

    def check_all(
        self, device_configs: List[Dict]
    ) -> Dict[str, ComplianceResult]:
        """对所有设备执行合规检查"""
        results = {}
        for config in device_configs:
            hostname = config.get("hostname", "unknown")
            results[hostname] = self.check_device(config)
        return results

四、合规报告生成

4.1 HTML 报告

#!/usr/bin/env python3
# compliance_report.py — 合规报告生成

from typing import Dict, List
import json


class ComplianceReportGenerator:
    """合规报告生成器"""

    @staticmethod
    def generate_html_report(
        results: Dict[str, ComplianceResult],
        rules: List[ComplianceRule],
        output_file: str = "compliance_report.html",
    ):
        """生成 HTML 格式合规报告"""

        # 计算全网合规率
        total_checks = sum(
            r.summary["pass"] + r.summary["fail"]
            for r in results.values()
        )
        total_pass = sum(
            r.summary["pass"] for r in results.values()
        )
        total_fail = sum(
            r.summary["fail"] for r in results.values()
        )
        pass_rate = (
            total_pass / total_checks * 100
            if total_checks > 0 else 0
        )

        # 构建规则统计
        rule_stats = {}
        for rule in rules:
            rule_stats[rule.id] = {
                "name": rule.name,
                "severity": rule.severity.value,
                "pass": 0,
                "fail": 0,
                "devices_failed": [],
            }
        for hostname, result in results.items():
            for check in result.results:
                if check.rule_id in rule_stats:
                    if check.passed:
                        rule_stats[check.rule_id]["pass"] += 1
                    else:
                        rule_stats[check.rule_id]["fail"] += 1
                        rule_stats[check.rule_id][
                            "devices_failed"
                        ].append(hostname)

        # 生成报告
        html = f"""
<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <title>网络配置合规性检查报告</title>
    <style>
        body {{ font-family: Arial, sans-serif; margin: 20px; }}
        .summary {{ background: #f0f8ff; padding: 20px; border-radius: 8px; }}
        .pass {{ color: green; }} .fail {{ color: red; }}
        table {{ border-collapse: collapse; width: 100%; }}
        th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
        th {{ background: #4CAF50; color: white; }}
        .critical {{ background: #ffebee; }}
        .high {{ background: #fff3e0; }}
        .medium {{ background: #fff8e1; }}
    </style>
</head>
<body>
    <h1>网络配置合规性检查报告</h1>
    <div class="summary">
        <h2>全网合规概览</h2>
        <p>检查时间: {results[next(iter(results))].timestamp if results else 'N/A'}</p>
        <p>检查设备数: {len(results)}</p>
        <p>检查项总数: {total_checks}</p>
        <p>通过: <span class="pass">{total_pass}</span></p>
        <p>未通过: <span class="fail">{total_fail}</span></p>
        <h3>全网合规率: {pass_rate:.1f}%</h3>
    </div>

    <h2>规则详细结果</h2>
    <table>
        <tr>
            <th>规则 ID</th>
            <th>规则名称</th>
            <th>严重级别</th>
            <th>通过</th>
            <th>未通过</th>
            <th>不合规设备</th>
        </tr>
"""

        for rule_id, stat in rule_stats.items():
            severity_class = ""
            if stat["severity"] == "CRITICAL":
                severity_class = "critical"
            elif stat["severity"] == "HIGH":
                severity_class = "high"
            elif stat["severity"] == "MEDIUM":
                severity_class = "medium"

            devices_failed = ", ".join(stat["devices_failed"])
            html += f"""
        <tr class="{severity_class}">
            <td>{rule_id}</td>
            <td>{stat['name']}</td>
            <td>{stat['severity']}</td>
            <td>{stat['pass']}</td>
            <td>{stat['fail']}</td>
            <td>{devices_failed}</td>
        </tr>
"""

        html += """
    </table>
</body>
</html>
"""
        with open(output_file, "w", encoding="utf-8") as f:
            f.write(html)
        print(f"合规报告已生成: {output_file}")

4.2 主流程

#!/usr/bin/env python3
# run_compliance_check.py — 合规检查主程序

from compliance_engine import DeviceConfigCollector, ComplianceChecker
from compliance_rules import ALL_RULES
from compliance_report import ComplianceReportGenerator
import json
import yaml


def main():
    # 加载设备清单
    with open("inventory.yml", "r") as f:
        inventory = yaml.safe_load(f)

    devices = inventory.get("devices", [])

    # 步骤 1：采集配置
    print(f"开始采集 {len(devices)} 台设备配置...")
    collector = DeviceConfigCollector()
    configs = collector.collect_all(devices)
    print(f"采集完成，成功: {len(configs)} 台")

    # 保存原始配置
    with open("collected_configs.json", "w") as f:
        # 可序列化处理
        json.dump(configs, f, default=str, indent=2)

    # 步骤 2：合规检查
    print("执行合规检查...")
    checker = ComplianceChecker(ALL_RULES)
    results = checker.check_all(configs)

    # 步骤 3：生成报告
    print("生成合规报告...")
    generator = ComplianceReportGenerator()
    generator.generate_html_report(results, ALL_RULES)

    # 输出总结
    total_fail = sum(
        r.summary["fail"] for r in results.values()
    )
    print(f"检查完成！发现 {total_fail} 个不合规项")

    if total_fail > 0:
        print("\n不合规设备详情：")
        for hostname, result in results.items():
            fails = [
                r for r in result.results if not r.passed
            ]
            if fails:
                print(f"\n{hostname}:")
                for f in fails:
                    print(f"  [{f.rule_id}] {f.detail}")


if __name__ == "__main__":
    main()

五、配置漂移检测

5.1 漂移检测原理

配置漂移检测：

  基线（Baseline）：
  ┌──────────────────────────────────────────┐
  │  采集时间：2024-01-01 00:00:00            │
  │  设备：CORE-SW01                          │
  │  ┌─ ssh version 2                        │
  │  ├─ aaa password encryption              │
  │  ├─ ntp server 10.0.0.1                 │
  │  └─ interface ...                        │
  └──────────────────────────────────────────┘

  当前配置（Current）：
  ┌──────────────────────────────────────────┐
  │  采集时间：2024-01-15 00:00:00            │
  │  设备：CORE-SW01                          │
  │  ┌─ ssh version 2                        │   ← 未变
  │  ├─ aaa password encryption              │   ← 未变
  │  ├─ ntp server 10.0.0.2                 │   ← 已变更!
  │  └─ interface ...                        │
  └──────────────────────────────────────────┘

  漂移检测结果：
  ┌──────────────────────────────────────────┐
  │  变更项：ntp server                       │
  │  ┌─ 基线值：10.0.0.1                     │
  │  ├─ 当前值：10.0.0.2                     │
  │  ├─ 变更时间：2024-01-12 14:23:45        │
  │  └─ 是否合规：需要重新评估                │
  └──────────────────────────────────────────┘

5.2 漂移检测实现

#!/usr/bin/env python3
# drift_detection.py — 配置漂移检测

from typing import Dict, List, Optional
from datetime import datetime
import difflib
import json


class DriftDetector:
    """配置漂移检测器"""

    def __init__(self, baseline_dir: str = "baseline/"):
        self.baseline_dir = baseline_dir

    def detect_drift(
        self,
        device: str,
        baseline_config: str,
        current_config: str,
    ) -> Dict:
        """检测单台设备的配置漂移"""

        # 计算差异
        diff = list(difflib.unified_diff(
            baseline_config.splitlines(keepends=True),
            current_config.splitlines(keepends=True),
            fromfile=f"{device}_baseline",
            tofile=f"{device}_current",
            n=3,
        ))

        if not diff:
            return {
                "device": device,
                "drifted": False,
                "changes": [],
                "summary": "无配置漂移",
            }

        # 解析变更
        changes = self._parse_diff(diff)

        return {
            "device": device,
            "drifted": True,
            "detected_at": datetime.now().isoformat(),
            "changes": changes,
            "diff_lines": diff,
            "summary": f"发现 {len(changes)} 处配置漂移",
        }

    def _parse_diff(self, diff: List[str]) -> List[Dict]:
        """解析 diff 为结构化变更"""
        changes = []
        current_section = ""

        for line in diff:
            if line.startswith("@@"):
                # 解析变更位置
                current_section = line
            elif line.startswith("+") and not line.startswith("+++"):
                changes.append({
                    "type": "added",
                    "content": line[1:].strip(),
                    "section": current_section,
                })
            elif line.startswith("-") and not line.startswith("---"):
                changes.append({
                    "type": "removed",
                    "content": line[1:].strip(),
                    "section": current_section,
                })

        return changes

    def detect_all(
        self,
        baselines: Dict[str, str],
        currents: Dict[str, str],
    ) -> Dict[str, Dict]:
        """批量检测所有设备的配置漂移"""
        results = {}
        for device in set(list(baselines.keys()) + list(currents.keys())):
            baseline = baselines.get(device, "")
            current = currents.get(device, "")
            results[device] = self.detect_drift(
                device, baseline, current
            )
        return results

六、最佳实践总结

自动化合规检查最佳实践：

  1. 规则管理
  ┌─ 规则存储在 Git 仓库，版本化管理
  ├─ 规则需定期评审和更新
  ├─ 区分硬性规则（必须遵守）和建议规则
  └─ 允许设备级别的规则豁免（需审批）

  2. 检查频率
  ┌─ 关键规则：每小时/每天
  ├─ 一般规则：每周
  ├─ 全量规则：每月
  └─ 变更后：立即触发检查

  3. 告警分级
  ┌─ CRITICAL：立即告警（电话/短信）
  ├─ HIGH：当天处理（邮件/IM）
  ├─ MEDIUM：纳入整改计划
  └─ LOW：月报汇总

  4. 持续改进
  ┌─ 从故障事件中提取新规则
  ├─ 合规报告与改进计划关联
  ├─ 定期回顾合规趋势
  └─ 自动化修复常见不合规项

  5. 与变更流程集成
  ┌─ 变更前：检查是否可能导致不合规
  ├─ 变更后：自动触发全量检查
  └─ 配置漂移：自动告警并记录

下篇预告：第324篇 — 自动化巡检与智能报告生成，讲解自动化巡检和智能报告生成的方法与工具。

下篇预告：第324篇 — 自动化巡检与智能报告生成，讲解自动化巡检和智能报告生成的方法与工具。