网络自动化：使用 Python 和 Ansible 自动化网络运维

网络自动化是现代网络工程师的必备技能。本文将介绍如何使用 Python 和 Ansible 等工具实现网络设备的自动化配置和管理。

网络自动化概述

为什么需要网络自动化？

传统手动操作                              自动化操作
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

┌──────────────┐                    ┌──────────────┐
│  人工 SSH    │                    │  批量执行    │
│   逐台登录   │         ===>        │  并行处理    │
└──────┬───────┘                    └──────┬───────┘
       │                                   │
       ▼                                   ▼
┌──────────────┐                    ┌──────────────┐
│  手动配置    │                    │  配置模板化   │
│  易出错      │         ===>        │  一致性高    │
└──────┬───────┘                    └──────┬───────┘
       │                                   │
       ▼                                   ▼
┌──────────────┐                    ┌──────────────┐
│  文档记录    │                    │ 版本控制     │
│  不完整      │         ===>        │ 可追溯       │
└──────────────┘                    └──────────────┘

优势：效率提升 10x+，错误率降低 90%+

自动化工具生态

类别	工具	适用场景
配置管理	Ansible, Chef, Puppet	配置同步、批量修改
编程接口	Netmiko, NAPALM, PyeAPI	编程控制、定制化
编排平台	AWX, Tower, Terraform	工作流编排、基础设施
监控自动化	Ansible Tower, RunDeck	任务调度、审批流程

Python 网络自动化

Netmiko 基础

#!/usr/bin/env python3
"""
Netmiko 基础用法示例
"""
from netmiko import ConnectHandler
from datetime import datetime

def backup_device(device_info):
    """备份网络设备配置"""
    print(f"连接到设备: {device_info['host']}")
    
    # 建立连接
    connection = ConnectHandler(**device_info)
    
    # 获取主机名
    prompt = connection.find_prompt()
    hostname = prompt.replace('#', '').replace('>', '')
    
    # 获取配置
    print("正在获取配置...")
    config = connection.send_command("show running-config")
    
    # 保存到文件
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    filename = f"backup_{hostname}_{timestamp}.cfg"
    
    with open(filename, 'w') as f:
        f.write(config)
    
    print(f"配置已保存到: {filename}")
    
    connection.disconnect()
    return filename

# 设备连接信息
device = {
    'device_type': 'cisco_ios',
    'host': '192.168.1.1',
    'username': 'admin',
    'password': 'cisco123',
    'secret': 'cisco123',  # 特权模式密码
}

# 执行备份
backup_device(device)

批量配置

#!/usr/bin/env python3
"""
批量配置网络设备
"""
from netmiko import ConnectHandler, exceptions
from concurrent.futures import ThreadPoolExecutor, as_completed
import logging

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

def configure_device(device_info, commands):
    """配置单个设备"""
    try:
        logger.info(f"连接 {device_info['host']}...")
        connection = ConnectHandler(**device_info)
        
        # 进入特权模式
        connection.enable()
        
        # 批量发送命令
        output = connection.send_config_set(commands)
        
        # 保存配置
        connection.save_config()
        
        logger.info(f"{device_info['host']} 配置完成")
        
        connection.disconnect()
        return {"status": "success", "host": device_info['host']}
        
    except exceptions.NetmikoAuthenticationException:
        logger.error(f"{device_info['host']} 认证失败")
        return {"status": "auth_error", "host": device_info['host']}
    except Exception as e:
        logger.error(f"{device_info['host']} 配置失败: {e}")
        return {"status": "error", "host": device_info['host'], "msg": str(e)}

def batch_config(devices, commands, max_workers=10):
    """批量配置多台设备"""
    results = []
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {
            executor.submit(configure_device, device, commands): device
            for device in devices
        }
        
        for future in as_completed(futures):
            result = future.result()
            results.append(result)
    
    return results

# 配置命令列表
config_commands = [
    'vlan 100',
    'name DATA_VLAN',
    'vlan 200',
    'name VOICE_VLAN',
    'interface range GigabitEthernet0/0 - 3',
    'switchport mode access',
    'switchport access vlan 100',
]

# 设备列表
devices = [
    {
        'device_type': 'cisco_ios',
        'host': '192.168.1.1',
        'username': 'admin',
        'password': 'cisco123',
        'secret': 'cisco123',
    },
    {
        'device_type': 'cisco_ios',
        'host': '192.168.1.2',
        'username': 'admin',
        'password': 'cisco123',
        'secret': 'cisco123',
    },
]

# 执行批量配置
results = batch_config(devices, config_commands)
print(results)

NAPALM 高级操作

#!/usr/bin/env python3
"""
使用 NAPALM 进行跨厂商网络自动化
"""
from napalm import get_network_driver

def get_device_info(host, driver_type):
    """获取设备信息"""
    drivers = {
        'cisco': 'ios',
        'huawei': 'vrp',
        'juniper': 'junos',
        'arista': 'eos',
    }
    
    driver_name = drivers.get(driver_type, 'ios')
    driver = get_network_driver(driver_name)
    
    device = driver(
        hostname=host,
        username='admin',
        password='cisco123',
        optional_args={'secret': 'cisco123'}
    )
    
    return device

def inventory_network(device_type, hosts):
    """批量获取网络设备信息"""
    results = []
    
    for host in hosts:
        try:
            device = get_device_info(host, device_type)
            device.open()
            
            # 获取信息
            facts = device.get_facts()
            interfaces = device.get_interfaces()
            arp_table = device.get_arp_table()
            
            results.append({
                'host': host,
                'status': 'success',
                'facts': facts,
                'interfaces_count': len(interfaces),
                'arp_entries': len(arp_table),
            })
            
            device.close()
            
        except Exception as e:
            results.append({
                'host': host,
                'status': 'error',
                'message': str(e),
            })
    
    return results

Ansible 网络自动化

Inventory 配置

# inventory.yml
all:
  children:
    cisco_devices:
      children:
        cisco_ios:
          hosts:
            core-sw-01:
              ansible_host: 192.168.1.1
              ansible_user: admin
              ansible_password: cisco123
            core-sw-02:
              ansible_host: 192.168.1.2
              ansible_user: admin
              ansible_password: cisco123
        cisco_nxos:
          hosts:
            nxos-switch:
              ansible_host: 192.168.2.1
              ansible_user: admin
              ansible_password: cisco123

    huawei_devices:
      hosts:
        huawei-sw-01:
          ansible_host: 192.168.3.1
          ansible_user: admin
          ansible_password: huawei123

Playbook 示例

# cisco_vlan_config.yml
---
- name: Configure VLANs on Cisco Devices
  hosts: cisco_ios
  gather_facts: no
  connection: local
  
  vars:
    vlans:
      - id: 10
        name: DATA
      - id: 20
        name: VOICE
      - id: 30
        name: WIRELESS
      - id: 99
        name: MGMT

  tasks:
    - name: Configure VLANs
      cisco.ios.ios_vlans:
        config: "{{ vlans }}"
        state: merged

    - name: Save configuration
      cisco.ios.ios_config:
        save_when: modified

角色结构

# Ansible 角色目录结构
roles/
└── network_vlan/
    ├── defaults/
    │   └── main.yml
    ├── tasks/
    │   └── main.yml
    ├── templates/
    │   └── vlan_config.j2
    └── handlers/
        └── main.yml

# roles/network_vlan/tasks/main.yml
---
- name: Deploy VLAN configuration
  cisco.ios.ios_vlans:
    config: "{{ vlans }}"
    state: "{{ vlans_state }}"

- name: Deploy port configurations
  cisco.ios.ios_l2_interfaces:
    config: "{{ port_configs }}"
    state: "{{ port_state }}"

Nornir 自动化框架

#!/usr/bin/env python3
"""
使用 Nornir 进行网络自动化
"""
from nornir import InitNornir
from nornir.core import Task
from nornir_utils.plugins.functions import print_result
from nornir_netmiko.tasks import netmiko_send_command, netmiko_send_config

def backup_config(task: Task) -> None:
    """备份配置"""
    task.run(
        task=netmiko_send_command,
        command_string="show running-config"
    )

def configure_vlan(task: Task, vlan_id: int, vlan_name: str) -> None:
    """配置 VLAN"""
    task.run(
        task=netmiko_send_config,
        config_commands=[
            f"vlan {vlan_id}",
            f"name {vlan_name}",
        ]
    )

# 初始化 Nornir
nr = InitNornir(config_file="nornir.yml")

# 备份所有设备配置
print("=== 备份配置 ===")
result = nr.run(task=backup_config)
print_result(result)

# 批量配置 VLAN
print("\n=== 配置 VLAN ===")
nr.run(
    task=configure_vlan,
    vlan_id=100,
    vlan_name="CORP_VLAN"
)

网络自动化最佳实践

配置模板化

{# templates/cisco_interface.j2 #}
interface {{ interface_name }}
 description {{ description }}
 {% if ip_address %}
 ip address {{ ip_address }} {{ subnet_mask }}
 {% endif %}
 {% if vlan %}
 switchport access vlan {{ vlan }}
 {% endif %}
 {% if switchport_mode %}
 switchport mode {{ switchport_mode }}
 {% endif %}
 no shutdown
!

版本控制

# Git 仓库结构
network-automation/
├── configs/
│   ├── templates/
│   │   └── interface.j2
│   └── variables/
│       └── site-a.yml
├── playbooks/
│   ├── deploy.yml
│   └── rollback.yml
├── roles/
│   └── network_device/
├── inventory/
│   ├── hosts.yml
│   └── group_vars/
├── tests/
│   └── test_configs.py
├── Jenkinsfile
└── README.md

CI/CD 流程

┌─────────┐     ┌─────────┐     ┌─────────┐     ┌─────────┐
│  Code   │────▶│  Build  │────▶│  Test   │────▶│ Deploy  │
│ Commit  │     │  & Lint │     │ & Valid │     │         │
└─────────┘     └─────────┘     └─────────┘     └─────────┘
                                    │
                                    ▼
                             ┌─────────────┐
                             │  Rollback   │
                             │  if Failed  │
                             └─────────────┘

总结

网络自动化是网络工程师职业发展的必备技能。建议从 Python 基础开始，逐步学习 Ansible，最终建立完整的自动化运维体系。通过自动化，您可以大幅提升工作效率，减少人为错误，为职业发展增添竞争力。