第324篇：Git 版本控制管理网络配置

关键词：Git、版本控制、分支策略、Merge Request、配置审计、回滚、Tag、网络配置管理

一、引言

Git 是软件工程中最重要的工具之一，它解决了代码的版本管理、团队协作和变更追溯问题。当网络配置被纳入代码化管理（NaC，见第323篇）后，Git 自然成为网络配置版本控制的首选工具。

将 Git 引入网络配置管理的核心价值：

• 变更可追溯：每一次配置修改都有明确的操作人、时间和变更内容
• 分支管理：变更可在隔离分支中进行，不影响现网运行配置
• 协作评审：通过 Merge Request（MR）实现配置变更的多层审核
• 一键回滚：任何历史版本的配置均可快速恢复

二、Git 仓库设计与目录规范

2.1 推荐仓库结构

network-configs/
- .gitignore
- CHANGELOG.md
- CONTRIBUTING.md
- PRODUCTION.md
- branches/
  - strategy.md
- configs/
  - production/           # 生产环境配置
    - pe-routers/
      - PE-BJ-01.yaml
      - PE-BJ-02.yaml
      - _golden.yaml       # 黄金配置基线
    - core-routers/
    - access-switches/
  - staging/              # 测试环境配置
  - templates/            # 通用配置模板
    - bgp_base.yaml
    - ospf_base.yaml
    - interface_base.yaml
- changes/                  # 变更请求目录
  - 20260901-bgp-timer-adjust/
    - description.md
    - PE-BJ-01.diff
    - rollback-plan.md
  - 20260905-acl-update/
- backups/                  # .gitignore 排除，仅本地使用
- scripts/
  - sync-from-devices.py  # 从设备同步配置
  - validate-all.py       # 批量验证
  - deploy-to-devices.py  # 部署到设备
- tests/
  - test_bgp_config.py
  - test_vlan_config.py

2.2 .gitignore 配置

# .gitignore
# 不提交敏感信息
*.password
*.secret
secrets/

# 不提交设备运行配置备份（太大且频繁变化）
backups/

# 不提交临时文件
*.tmp
*.bak
__pycache__/
*.pyc

# 不提交 IDE 配置
.vscode/
.idea/

# 但提交变更请求中的 diff
# changes/**/*.diff  - 允许

三、分支策略与变更流程

3.1 Git Flow 分支模型

flowchart LR
    main[(main)] --> hotfix["hotfix/*"]
    main --> release["release/*"]
    develop[(develop)] --> feature["feature/*"]
    develop --> release

    style main fill:#C8E6C9
    style develop fill:#E3F2FD
    style feature fill:#FFF3E0
    style release fill:#F3E5F5
    style hotfix fill:#FFEBEE

3.2 标准变更流程

sequenceDiagram
    participant Eng as 工程师
    participant Git as Git 仓库
    participant Reviewer as 评审人
    participant CI as CI 检查

    Eng->>Git: git checkout -b feature/vlan-update
    Eng->>Git: 修改配置文件
    Eng->>Git: git commit + push
    Git->>CI: 触发自动化检查
    CI-->>Git: 检查结果（Lint/Test）
    Eng->>Git: 创建 Merge Request
    Git->>Reviewer: 通知评审
    Reviewer->>Git: 审查 + 评论
    Reviewer->>Git: 批准 MR
    Git->>Eng: MR 合并到 develop
    Eng->>Git: 部署到 staging 验证
    Eng->>Git: 创建 Release MR
    Reviewer->>Git: 最终批准
    Git->>Git: 合并到 main

3.3 实用 Git 命令

# 1. 创建变更分支
git checkout develop
git checkout -b feature/20260901-bgp-timer-adjust

# 2. 修改配置并提交（每次提交应独立、可验证）
git add configs/production/pe-routers/PE-BJ-01.yaml
git commit -m "feat(bgp): 调整 PE-BJ-01 BGP keepalive timer 至 10s

- keepalive 从 30s 调整为 10s
- holdtime 从 90s 调整为 30s
- 变更原因: 加快故障检测速度
- 关联工单: CHG-20260901-001"

# 3. 推送到远端并创建 MR
git push origin feature/20260901-bgp-timer-adjust

# 4. 查看变更内容
git diff main...feature/20260901-bgp-timer-adjust

# 5. 紧急回滚
git revert HEAD --no-edit      # 回滚最近一次提交
git revert abc1234 --no-edit   # 回滚指定提交

四、配置审计与追溯

4.1 Git Blame 定位变更责任人

# 查看某行配置是谁、何时修改的
git blame configs/production/pe-routers/PE-BJ-01.yaml

# 输出示例:
# ^7a3b2c1 (Zhang San  2026-08-15 14:30:22 +0800 12) peer 10.1.1.2 as-number 65002
# d4e5f6a7 (Li Si     2026-09-01 10:15:33 +0800 13)  peer 10.1.1.2 timer 10 30

4.2 自动化配置审计脚本

# scripts/config_audit.py
import subprocess
import json
from datetime import datetime

def audit_config_history(config_file, since='2026-01-01'):
    """审计配置文件的变更历史"""
    # 获取日志
    log = subprocess.run(
        ['git', 'log', '--since', since,
         '--format=%H|%an|%ae|%ai|%s',
         '--', config_file],
        capture_output=True, text=True
    ).stdout

    changes = []
    for line in log.strip().split('\n'):
        if not line:
            continue
        parts = line.split('|', 4)
        changes.append({
            'commit': parts[0][:8],
            'author': parts[1],
            'email': parts[2],
            'date': parts[3],
            'message': parts[4]
        })

    # 统计
    authors = {}
    for c in changes:
        authors[c['author']] = authors.get(c['author'], 0) + 1

    return {
        'file': config_file,
        'total_changes': len(changes),
        'first_change': changes[-1]['date'] if changes else None,
        'last_change': changes[0]['date'] if changes else None,
        'contributors': authors,
        'changes': changes
    }

def generate_audit_report(config_dir):
    """生成批量审计报告"""
    import os
    report = []
    for root, dirs, files in os.walk(config_dir):
        for f in files:
            if f.endswith('.yaml'):
                filepath = os.path.join(root, f)
                audit = audit_config_history(filepath)
                report.append(audit)

    # 输出 JSON
    with open('audit_report.json', 'w') as f:
        json.dump(report, f, indent=2, ensure_ascii=False)

    return report

# 运行审计
report = generate_audit_report('configs/production/')
print(f"审计了 {len(report)} 个配置文件")
for r in report:
    print(f"  {r['file']}: {r['total_changes']} 次变更, "
          f"{len(r['contributors'])} 个贡献者")

4.3 Tag 管理发布版本

# 为每个生产发布打 Tag
git tag -a v2026.09.01 -m "Release 2026-09-01: BGP timer 调整 + ACL 更新"
git push origin v2026.09.01

# 查看 Tag 历史
git tag -l --sort=-creatordate

# 回滚到指定 Tag
git checkout v2026.08.15
python scripts/deploy-to-devices.py --env production --tag v2026.08.15

五、Git Hooks 自动化验证

在提交或推送前自动执行验证，防止错误配置进入仓库。

5.1 Pre-commit Hook：提交前自动检查

#!/bin/bash
# .git/hooks/pre-commit — 提交前自动验证

echo "=== Pre-commit 验证 ==="

# 检查 YAML 格式
echo "1. YAML 语法检查..."
for file in $(git diff --cached --name-only --diff-filter=AM | grep -E '\.(yml|yaml)$'); do
    python -c "import yaml; yaml.safe_load(open('$file'))" 2>/dev/null
    if [ $? -ne 0 ]; then
        echo "❌ $file: YAML 格式错误"
        exit 1
    fi
    echo "   ✓ $file"
done

# 检查 JSON 格式
echo "2. JSON 语法检查..."
for file in $(git diff --cached --name-only --diff-filter=AM | grep '\.json$'); do
    python -c "import json; json.load(open('$file'))" 2>/dev/null
    if [ $? -ne 0 ]; then
        echo "❌ $file: JSON 格式错误"
        exit 1
    fi
    echo "   ✓ $file"
done

# 检查 Python 语法
echo "3. Python 语法检查..."
for file in $(git diff --cached --name-only --diff-filter=AM | grep '\.py$'); do
    python -m py_compile "$file" 2>/dev/null
    if [ $? -ne 0 ]; then
        echo "❌ $file: Python 语法错误"
        exit 1
    fi
    echo "   ✓ $file"
done

# 检查是否包含明文密码
echo "4. 敏感信息检查..."
for file in $(git diff --cached --name-only --diff-filter=AM); do
    if grep -q -E '(password|secret|passwd)\s*[:=]\s*[^$]' "$file" 2>/dev/null; then
        if ! echo "$file" | grep -q '\.vault'; then
            echo "⚠  $file: 可能包含明文密码"
        fi
    fi
done

echo "=== Pre-commit 验证通过 ==="

5.2 Post-merge Hook：合并后同步配置

#!/usr/bin/env python3
# .git/hooks/post-merge — 合并后自动检查配置变更
import subprocess
import sys
import json

def get_changed_files(is_merge):
    """获取本次合并涉及的文件"""
    if is_merge:
        cmd = ['git', 'diff-tree', '--no-commit-id', '--name-only', '-r', 'HEAD']
    else:
        cmd = ['git', 'diff', '--name-only', 'HEAD~1', 'HEAD']
    result = subprocess.run(cmd, capture_output=True, text=True)
    return [f for f in result.stdout.strip().split('\n') if f]

def notify_config_change(changed_files):
    """通知配置变更"""
    config_files = [f for f in changed_files if f.endswith('.yaml') or f.endswith('.j2')]
    if config_files:
        msg = f"⚠ 配置变更提醒: {len(config_files)} 个配置文件被修改\n"
        for f in config_files:
            msg += f"  - {f}\n"
        msg += "\n请及时部署到对应设备！"
        print(msg)
        # 可选：通过 Webhook 发送到飞书/钉钉
        # send_webhook(msg)

if __name__ == '__main__':
    is_merge = len(sys.argv) > 1 and sys.argv[1] == '1'
    changed = get_changed_files(is_merge)
    notify_config_change(changed)

六、Python Git 操作封装

为自动化流程提供统一的 Git 操作接口。

#!/usr/bin/env python3
# scripts/git_manager.py — Git 仓库管理器

import subprocess
import os
from typing import List, Optional
from datetime import datetime

class GitManager:
    """Git 仓库管理器"""

    def __init__(self, repo_path: str = "."):
        self.repo_path = os.path.abspath(repo_path)

    def _run_git(self, args: List[str]) -> str:
        """执行 Git 命令"""
        cmd = ["git"] + args
        result = subprocess.run(
            cmd, capture_output=True, text=True, cwd=self.repo_path
        )
        if result.returncode != 0:
            raise RuntimeError(
                f"Git 命令失败: {' '.join(cmd)}\n{result.stderr}"
            )
        return result.stdout.strip()

    @property
    def current_branch(self) -> str:
        """当前分支名"""
        return self._run_git(["rev-parse", "--abbrev-ref", "HEAD"])

    @property
    def last_commit(self) -> str:
        """最后一次提交信息"""
        return self._run_git(["log", "-1", "--pretty=format:%h %s"])

    def status(self) -> str:
        """查看仓库状态"""
        return self._run_git(["status"])

    def diff(self, since: str = "HEAD~1") -> str:
        """查看最近变更"""
        return self._run_git(["diff", since, "--stat"])

    def changed_files(self, since: str = "HEAD~1") -> List[str]:
        """获取变更文件列表"""
        output = self._run_git(["diff", since, "--name-only"])
        return [f for f in output.split("\n") if f]

    def commit_and_push(self, message: str, branch: str = None) -> bool:
        """提交并推送"""
        try:
            self._run_git(["add", "-A"])
            status = self._run_git(["status", "--porcelain"])
            if not status:
                print("无变更需要提交")
                return False
            self._run_git(["commit", "-m", message])
            commit_hash = self._run_git(["rev-parse", "HEAD"])
            print(f"✓ 提交成功: {commit_hash[:8]} - {message}")
            target_branch = branch or self.current_branch
            self._run_git(["push", "-u", "origin", target_branch])
            print(f"✓ 已推送到 origin/{target_branch}")
            return True
        except RuntimeError as e:
            print(f"✗ Git 操作失败: {e}")
            return False

    def create_branch(self, branch_name: str, base: str = "develop"):
        """创建分支"""
        self._run_git(["checkout", base])
        self._run_git(["pull"])
        self._run_git(["checkout", "-b", branch_name])
        print(f"✓ 已创建分支: {branch_name} (基于 {base})")

    def get_commit_history(self, path: str = "", limit: int = 10) -> List[dict]:
        """获取提交历史"""
        args = [
            "log", f"--max-count={limit}",
            "--pretty=format:%H|%an|%ad|%s", "--date=iso",
        ]
        if path:
            args.extend(["--", path])
        output = self._run_git(args)
        commits = []
        for line in output.split("\n"):
            if not line:
                continue
            parts = line.split("|", 3)
            if len(parts) == 4:
                commits.append({
                    "hash": parts[0][:8],
                    "author": parts[1],
                    "date": parts[2],
                    "message": parts[3],
                })
        return commits

    def tag_release(self, version: str, message: str = ""):
        """创建发布标签"""
        msg = message or f"Release {version}"
        self._run_git(["tag", "-a", version, "-m", msg])
        self._run_git(["push", "origin", version])
        print(f"✓ 已创建标签: {version}")

    def get_config_diff(self, device_name: str) -> str:
        """获取特定设备的配置变更"""
        return self._run_git([
            "log", "-p", "--all", "--diff-filter=M",
            "--", f"*{device_name}*", "-n", "5",
        ])

七、MR 规范与提交信息标准

7.1 MR 模板

<!-- .gitlab/merge_request_templates/network_change.md -->

## 变更描述

**变更类型**: (新增/修改/删除/回滚)
**关联需求**: #123

## 变更范围

- [ ] 接入交换机
- [ ] 核心交换机
- [ ] 路由器
- [ ] 防火墙

## 变更内容

### 配置文件修改
- `config/vlans.yml`: 新增 VLAN 200
- `templates/access_switch.j2`: 模板更新

### 变更摘要
```diff
+ vlan 200
+  name DEV

验证结果

• [x] YAML 语法检查通过
• [x] 模板渲染验证通过
• [x] IP 冲突检测通过
• [x] 合规策略检查通过
• [ ] 测试环境验证通过

风险评估

• 影响范围: ACC-SW01 ~ ACC-SW05
• 回滚方案: git revert + 重新部署
• 变更窗口: 2026-01-15 02:00-04:00

审批人

• [ ] @network-lead
• [ ] @change-manager

### 7.2 提交信息规范

Commit Message 格式：

():

类型说明: - feat: 新功能（新增 VLAN/接口/BGP 配置） - fix: 修复（配置错误/Bug 修复） - config: 配置变更（参数调整） - refactor: 重构（模板优化/目录重组） - test: 测试（测试用例新增/修改） - docs: 文档（README/注释更新） - chore: 杂项（CI 配置/依赖更新）

实际示例: feat(vlan): 新增 VLAN 200 用于开发环境 fix(bgp): 修复邻居 10.0.0.1 的 AS 号配置错误 config(ntp): 更新 NTP 服务器 IP 地址 chore(ci): 添加模板验证 Job refactor(templates): 抽取宏定义到 macros.j2

## 八、配置审计与合规检查

### 8.1 自动化审计报告

```python
#!/usr/bin/env python3
# scripts/audit_check.py — Git 审计检查

from git_manager import GitManager
import re

class GitAudit:
    """Git 审计检查器"""

    def __init__(self, repo_path: str):
        self.git = GitManager(repo_path)

    def check_commit_messages(self, limit: int = 100) -> list:
        """检查提交信息是否符合规范"""
        violations = []
        pattern = r'^(feat|fix|config|refactor|test|docs|chore)\([^)]+\): .+'
        for c in self.git.get_commit_history(limit=limit):
            if not re.match(pattern, c["message"]):
                violations.append(c)
        return violations

    def check_sensitive_files(self, since: str = "HEAD~10") -> list:
        """检查是否包含敏感文件"""
        sensitive_patterns = [r'.*\.pem$', r'.*\.key$',
                             r'.*password.*', r'.*secret.*']
        violations = []
        changed = self.git.changed_files(since=since)
        for f in changed:
            for pattern in sensitive_patterns:
                if re.match(pattern, f):
                    violations.append(f)
        return violations

    def generate_report(self) -> str:
        """生成审计报告"""
        report = []
        report.append("=" * 50)
        report.append("Git 审计报告")
        report.append(f"生成时间: {datetime.now()}")
        report.append("=" * 50)

        commits = self.git.get_commit_history(limit=500)
        authors = {}
        for c in commits:
            authors[c["author"]] = authors.get(c["author"], 0) + 1

        report.append(f"\n提交总数: {len(commits)}")
        report.append("\n贡献者统计:")
        for author, count in sorted(authors.items(), key=lambda x: -x[1]):
            report.append(f"  {author}: {count} 次提交")

        bad_msgs = self.check_commit_messages()
        if bad_msgs:
            report.append(f"\n⚠ 不合规的提交信息 ({len(bad_msgs)} 条):")
            for c in bad_msgs[:10]:
                report.append(f"  {c['hash']}: {c['message']}")

        sensitive = self.check_sensitive_files()
        if sensitive:
            report.append(f"\n❌ 可疑的敏感文件 ({len(sensitive)} 个):")
            for f in sensitive:
                report.append(f"  {f}")

        return "\n".join(report)

if __name__ == "__main__":
    audit = GitAudit(".")
    print(audit.generate_report())

八（续）、配置备份完整代码与 Git 集成方案

8.1 配置备份完整实现

#!/usr/bin/env python3
# scripts/config_backup_complete.py
"""
完整的配置备份方案：采集 → Git 提交 → 差异分析
"""
import os
import hashlib
import difflib
import logging
import subprocess
from pathlib import Path
from datetime import datetime
from netmiko import ConnectHandler
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import Dict, List, Optional

logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
logger = logging.getLogger(__name__)


class GitConfigBackup:
    """配置备份与 Git 集成的完整实现"""

    def __init__(self, git_repo: str = './network-configs',
                 credentials: Dict = None,
                 ssh_key: str = None):
        self.git_repo = Path(git_repo)
        self.credentials = credentials or {}
        self.ssh_key = ssh_key
        # 上次备份的哈希
        self.prev_hashes: Dict[str, str] = {}
        self._load_prev_hashes()

    def _load_prev_hashes(self):
        hash_file = self.git_repo / '.backup_hashes.json'
        if hash_file.exists():
            import json
            with open(hash_file) as f:
                self.prev_hashes = json.load(f)

    def _save_hashes(self):
        import json
        hash_file = self.git_repo / '.backup_hashes.json'
        with open(hash_file, 'w') as f:
            json.dump(self.prev_hashes, f, indent=2)

    def backup_device_to_git(self, device: Dict) -> Dict:
        """单设备备份并自动提交到 Git"""
        dev_name = device['name']
        result = {
            'device': dev_name,
            'status': 'success',
            'has_changes': False,
            'commit_hash': None,
            'errors': []
        }

        try:
            # 1. 连接设备并采集配置
            params = {
                'device_type': device.get('platform', 'huawei_vrpv8'),
                'host': device['ip'],
                'username': device.get('username', self.credentials.get('username')),
                'password': device.get('password', self.credentials.get('password')),
                'timeout': 30,
            }

            with ConnectHandler(**params) as net:
                running = net.send_command('display current-configuration')

            # 2. 计算当前配置的哈希
            current_hash = hashlib.sha256(running.encode()).hexdigest()

            # 3. 写入文件
            config_dir = self.git_repo / 'configs' / 'running'
            config_dir.mkdir(parents=True, exist_ok=True)
            config_file = config_dir / f"{dev_name}.cfg"
            config_file.write_text(running, encoding='utf-8')

            # 4. 检测变更
            prev_hash = self.prev_hashes.get(dev_name, '')
            has_changes = prev_hash and prev_hash != current_hash

            if has_changes:
                result['has_changes'] = True

                # 5. 生成差异文件
                diff_file = config_dir / f"{dev_name}.diff"
                if prev_hash:
                    diff = self._generate_diff(dev_name, running)
                    if diff:
                        diff_file.write_text(diff, encoding='utf-8')

                # 6. 生成摘要文件
                summary = self._generate_summary(running, dev_name)

            # 7. 更新哈希
            self.prev_hashes[dev_name] = current_hash

            # 8. Git 提交（如果有变更）
            if has_changes:
                try:
                    import subprocess
                    subprocess.run(['git', 'add', str(config_file),
                                   str(diff_file)],
                                   cwd=self.git_repo, check=True)
                    commit_msg = (f"backup: {dev_name} config changed\n"
                                  f"hash: {prev_hash[:12]} -> {current_hash[:12]}")
                    cp = subprocess.run(
                        ['git', 'commit', '-m', commit_msg],
                        cwd=self.git_repo,
                        capture_output=True, text=True
                    )
                    if cp.returncode == 0:
                        # 提取 commit hash
                        log = subprocess.run(
                            ['git', 'log', '-1', '--format=%h'],
                            cwd=self.git_repo, capture_output=True, text=True
                        )
                        result['commit_hash'] = log.stdout.strip()
                        logger.info(f"设备 {dev_name} 变更已提交: {result['commit_hash']}")
                except subprocess.CalledProcessError as e:
                    logger.warning(f"Git 提交失败: {e}")

        except Exception as e:
            result['status'] = 'failed'
            result['errors'].append(str(e))
            logger.error(f"设备 {dev_name} 备份失败: {e}")

        return result

    def _generate_diff(self, device_name: str, new_config: str) -> str:
        """生成配置差异"""
        config_file = self.git_repo / 'configs' / 'running' / f"{device_name}.cfg.prev"
        if config_file.exists():
            with open(config_file, 'r') as f:
                old_config = f.read()
            diff = difflib.unified_diff(
                old_config.splitlines(),
                new_config.splitlines(),
                fromfile='previous', tofile='current', lineterm=''
            )
            return '\n'.join(diff)
        return ""

    def _generate_summary(self, config: str, device_name: str) -> Dict:
        """生成配置摘要"""
        summary = {
            'device': device_name,
            'timestamp': datetime.now().isoformat(),
            'total_lines': len(config.splitlines()),
            'total_size': len(config.encode()),
        }
        # 统计 VLAN 数
        vlan_count = sum(1 for line in config.splitlines()
                        if line.strip().startswith('vlan '))
        summary['vlan_count'] = vlan_count
        # 统计接口数
        interface_count = sum(1 for line in config.splitlines()
                             if line.strip().startswith('interface '))
        summary['interface_count'] = interface_count
        return summary

    def backup_all(self, devices: List[Dict],
                   push_to_remote: bool = True) -> List[Dict]:
        """批量备份所有设备并推送到远程仓库"""
        results = []
        with ThreadPoolExecutor(max_workers=10) as executor:
            futures = {
                executor.submit(self.backup_device_to_git, d): d['name']
                for d in devices
            }
            for future in as_completed(futures):
                results.append(future.result())

        self._save_hashes()

        # 推送到远程
        if push_to_remote:
            try:
                subprocess.run(
                    ['git', 'push', 'origin', 'main'],
                    cwd=self.git_repo, check=True
                )
                logger.info("已推送到远程仓库")
            except subprocess.CalledProcessError as e:
                logger.error(f"推送失败: {e}")

        # 汇总
        changed = sum(1 for r in results if r['has_changes'])
        failed = sum(1 for r in results if r['status'] == 'failed')
        logger.info(f"备份完成: 总计{len(results)}台, "
                    f"变更{changed}台, 失败{failed}台")
        return results

8.2 Git 集成方案

# .gitlab-ci.yml - Git 集成 Pipeline
stages:
  - backup
  - validate
  - diff-report
  - notify

# 定时备份任务
scheduled_backup:
  stage: backup
  image: python:3.11-slim
  script:
    - pip install -r requirements.txt
    - python scripts/config_backup_complete.py --devices inventory/all_devices.yaml
  schedule:
    - cron: "0 */6 * * *"  # 每6小时
  rules:
    - if: $CI_PIPELINE_SOURCE == "schedule"

# 配置变更检测
config_diff_check:
  stage: validate
  image: python:3.11-slim
  script:
    - pip install -r requirements.txt
    - python scripts/config_diff_check.py
      --repo-dir ./network-configs
      --output reports/config_diff.json
  artifacts:
    paths:
      - reports/
    expire_in: 30 days
  rules:
    - if: $CI_PIPELINE_SOURCE == "schedule"

# 差异报告生成
diff_report:
  stage: diff-report
  image: python:3.11-slim
  script:
    - python scripts/generate_diff_report.py
      --input reports/config_diff.json
      --output reports/config_diff_report.html
  artifacts:
    paths:
      - reports/
    expire_in: 30 days
  needs:
    - config_diff_check

# 变更通知
notify_changes:
  stage: notify
  image: curlimages/curl:latest
  script:
    - |
      CHANGES=$(cat reports/config_diff.json | python -c "
      import sys, json
      d = json.load(sys.stdin)
      print(f'检测到 {d[\"changed_devices\"]} 台设备配置变更')
      ")
      curl -X POST "$FLYBOOK_WEBHOOK" \
        -H "Content-Type: application/json" \
        -d "{\"msg_type\":\"text\",\"content\":{\"text\":\"$CHANGES\"}}"
  rules:
    - if: $CI_PIPELINE_SOURCE == "schedule"

8.3 配置差异分析工具

#!/usr/bin/env python3
# scripts/config_diff_analyzer.py
"""
配置差异分析工具
功能：逐行 Diff、结构化分析、变更风险评估
"""
import difflib
import json
import yaml
from pathlib import Path
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass, field, asdict


@dataclass
class ConfigChange:
    """单条配置变更"""
    device: str
    line_number: int
    change_type: str  # added / removed / modified
    old_value: str = ""
    new_value: str = ""
    category: str = "unknown"  # vlan / interface / bgp / ospf / acl / system
    risk_level: str = "low"    # low / medium / high / critical


@dataclass
class DiffReport:
    """差异分析报告"""
    device: str
    timestamp: str
    total_changes: int
    additions: int
    deletions: int
    changes: List[Dict] = field(default_factory=list)
    risk_summary: Dict = field(default_factory=dict)


class ConfigDiffAnalyzer:
    """配置差异分析器"""

    # 分类规则
    CATEGORIES = {
        'vlan': [r'^vlan\s', r'^  name\s'],
        'interface': [r'^interface\s', r'^\s+ip\s+address', r'^\s+port\s'],
        'bgp': [r'^bgp\s', r'^\s+peer\s'],
        'ospf': [r'^ospf\s', r'^\s+area\s', r'^\s+network\s'],
        'acl': [r'^acl\s', r'^\s+rule\s', r'^traffic-filter\s'],
        'system': [r'^sysname\s', r'^snmp-agent\s', r'^ntp\s'],
    }

    @staticmethod
    def categorize_line(line: str) -> str:
        """将配置行分类"""
        for category, patterns in ConfigDiffAnalyzer.CATEGORIES.items():
            for pattern in patterns:
                import re
                if re.match(pattern, line.strip()):
                    return category
        return 'other'

    @staticmethod
    def assess_risk(change: ConfigChange) -> str:
        """评估变更风险等级"""
        if change.category in ('bgp', 'acl'):
            if change.change_type == 'removed':
                return 'critical'
            return 'high'
        if change.category == 'interface':
            if 'shutdown' in change.new_value.lower():
                return 'critical'
            if change.change_type == 'modified':
                return 'medium'
            return 'low'
        if change.category == 'vlan':
            return 'medium'
        return 'low'

    def compare_configs(self, old_config: str, new_config: str,
                        device_name: str = '') -> DiffReport:
        """比对两份配置文件"""
        old_lines = old_config.splitlines()
        new_lines = new_config.splitlines()

        diff = difflib.unified_diff(
            old_lines, new_lines,
            fromfile='old', tofile='new', lineterm=''
        )

        changes = []
        additions = 0
        deletions = 0

        for line in diff:
            if line.startswith('+') and not line.startswith('+++'):
                additions += 1
                change = ConfigChange(
                    device=device_name,
                    line_number=0,
                    change_type='added',
                    new_value=line[1:].strip(),
                    category=self.categorize_line(line)
                )
                change.risk_level = self.assess_risk(change)
                changes.append(asdict(change))

            elif line.startswith('-') and not line.startswith('---'):
                deletions += 1
                change = ConfigChange(
                    device=device_name,
                    line_number=0,
                    change_type='removed',
                    old_value=line[1:].strip(),
                    category=self.categorize_line(line)
                )
                change.risk_level = self.assess_risk(change)
                changes.append(asdict(change))

        # 风险汇总
        risk_counts = {'low': 0, 'medium': 0, 'high': 0, 'critical': 0}
        for c in changes:
            risk_counts[c['risk_level']] = risk_counts.get(c['risk_level'], 0) + 1

        return DiffReport(
            device=device_name,
            timestamp=datetime.now().isoformat(),
            total_changes=additions + deletions,
            additions=additions,
            deletions=deletions,
            changes=changes,
            risk_summary=risk_counts
        )

8.4 回滚机制实现

#!/usr/bin/env python3
# scripts/config_rollback.py
"""
配置回滚机制
支持：按 Tag 回滚、按 Commit 回滚、自动回滚到上次已知良好的状态
"""
import subprocess
import logging
from pathlib import Path
from typing import Dict, List, Optional
from datetime import datetime
from netmiko import ConnectHandler

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class ConfigRollback:
    """配置回滚管理器"""

    def __init__(self, git_repo: str, credentials: Dict):
        self.git_repo = Path(git_repo)
        self.credentials = credentials

    def rollback_to_tag(self, tag: str, devices: List[Dict] = None):
        """回滚到指定 Tag"""
        logger.info(f"回滚到 Tag: {tag}")

        # 1. 检出 Tag
        subprocess.run(
            ['git', 'checkout', tag],
            cwd=self.git_repo, check=True
        )

        # 2. 获取 Tag 对应的配置
        config_dir = self.git_repo / 'configs' / 'production'

        if devices:
            target_devices = devices
        else:
            # 获取所有配置文件的设备名
            target_devices = []
            for cfg_file in config_dir.glob('*.yaml'):
                target_devices.append({
                    'name': cfg_file.stem,
                    'config_file': str(cfg_file)
                })

        # 3. 逐台部署
        for device in target_devices:
            self._deploy_config(device)

        logger.info(f"回滚完成: 已回滚 {len(target_devices)} 台设备到 {tag}")

    def rollback_to_commit(self, commit_hash: str, device_name: str = None):
        """回滚到指定 Commit"""
        logger.info(f"回滚到 Commit: {commit_hash[:8]}")

        subprocess.run(
            ['git', 'checkout', commit_hash, '--',
             'configs/production/'],
            cwd=self.git_repo, check=True
        )

        if device_name:
            config_file = self.git_repo / 'configs' / 'production' / f"{device_name}.yaml"
            if config_file.exists():
                self._deploy_config({'name': device_name,
                                     'config_file': str(config_file)})

    def auto_rollback(self, device: Dict, backup_file: str):
        """自动回滚到备份文件"""
        logger.info(f"设备 {device['name']} 自动回滚")

        try:
            with ConnectHandler(
                device_type=device.get('platform', 'huawei_vrpv8'),
                host=device['ip'],
                username=device.get('username', self.credentials.get('username')),
                password=device.get('password', self.credentials.get('password')),
            ) as net:
                # 重置配置并重新加载备份
                net.send_command('reset saved-configuration', expect_string='Are you sure')
                net.send_command('Y')
                # 从备份文件加载配置
                backup_config = Path(backup_file).read_text(encoding='utf-8')
                for line in backup_config.splitlines():
                    if line.strip() and not line.strip().startswith('#'):
                        try:
                            net.send_command(line)
                        except:
                            pass  # 跳过无法直接应用的行
                net.send_command('save force')
                logger.info(f"设备 {device['name']} 回滚成功")
        except Exception as e:
            logger.error(f"设备 {device['name']} 回滚失败: {e}")
            raise

    def _deploy_config(self, device: Dict):
        """部署配置到设备"""
        if 'config_file' in device:
            import yaml
            with open(device['config_file']) as f:
                config_data = yaml.safe_load(f)
            # 生成 CLI 配置并下发
            cli_config = self._render_config(config_data)
            self._push_to_device(device, cli_config)
        logger.info(f"设备 {device['name']} 配置已部署")

    def _render_config(self, config_data: Dict) -> str:
        """将结构化配置渲染为 CLI"""
        from jinja2 import Template
        # 简化版，实际应使用 Jinja2 模板
        lines = []
        for section, data in config_data.items():
            if isinstance(data, dict):
                for key, value in data.items():
                    lines.append(f"  {key} {value}")
        return '\n'.join(lines)

    def _push_to_device(self, device: Dict, config: str):
        """推送配置到设备"""
        with ConnectHandler(
            device_type=device.get('platform', 'huawei_vrpv8'),
            host=device['ip'],
            username=device.get('username', self.credentials.get('username')),
            password=device.get('password', self.credentials.get('password')),
        ) as net:
            net.send_config_set(config.splitlines())
            net.save_config()

九、最佳实践与踩坑记录

9.1 最佳实践

9.2 常见踩坑

十、总结

故障案例回顾：某运营商 BGP 路由出现异常，核心路由器开始发布默认路由到所有邻居。通过 git blame 快速定位到 3 天前的问题提交，git show 查看变更上下文，git revert 即时回滚。该案例体现了 Git 在故障排查中的三个核心价值：快速定位（blame）、理解上下文（show）、即时回滚（revert）。

思考题

1. 在 Git 管理网络配置时，如何处理配置文件中的敏感信息（如 BGP MD5 密码、SNMP community）？有哪些安全的解决方案？
2. 当多个工程师同时修改同一台设备的配置时，Git 合并冲突如何解决？网络配置合并与代码合并有何不同？

下篇预告：第325篇《配置变更审批与回滚流程自动化》——构建完整的变更审批工作流，实现自动化回滚保障变更安全。