### Python 网络自动化运维实现方法与示例代码 #### 1. ### 利用 Paramiko 进行 SSH 配置管理 Paramiko 是一个强大的第三方库，支持通过 SSH 协议远程连接到网络设备并发送命令。以下是使用 Paramiko 执行路由器或交换机配置更新的一个简单示例[^1]。 ```python import paramiko def ssh_command(ip, user, passwd, command): client = paramiko.SSHClient() # 自动添加主机密钥策略 client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) try: client.connect(ip, username=user, password=passwd) _, stdout, stderr = client.exec_command(command) output = stdout.read().decode() + stderr.read().decode() return output except Exception as e: return str(e) finally: client.close() # 示例调用 ip_address = '192.168.1.1' username = 'admin' password = 'secret' command_to_execute = 'show ip interface brief' result = ssh_command(ip_address, username, password, command_to_execute) print(result) ``` 此脚本允许管理员轻松获取目标设备的状态信息或其他诊断数据。 --- #### 2. ### 使用 Netmiko 处理多种厂商设备 Netmiko 构建于 Paramiko 之上，提供了针对 Cisco、Juniper、HP ProCurve 等不同品牌硬件的支持。它简化了许多常见的任务流程，比如登录验证过程以及超时处理等问题[^1]。 安装 netmiko 库： ```bash pip install netmiko ``` 下面是利用 Netmiko 修改思科 IOS 路由器接口 IP 地址的例子： ```python from netmiko import ConnectHandler device_info = { 'device_type': 'cisco_ios', 'host': '192.168.1.1', 'username': 'admin', 'password': 'secret', } commands = [ 'configure terminal', 'interface GigabitEthernet0/1', 'no shutdown', 'ip address 172.16.1.1 255.255.255.0', 'exit', ] try: connection = ConnectHandler(**device_info) output = connection.send_config_set(commands) print(output) except Exception as error: print(f"Error occurred: {error}") finally: if 'connection' in locals(): connection.disconnect() ``` 这段程序展示了如何向指定端口分配新的 IPv4 参数集。 --- #### 3. ### Flask Web 接口配合 Celery 定期巡检 为了使整个系统更加友好易用，还可以搭建 RESTful API 来接受来自前端页面的操作请求。与此同时，借助消息队列工具 Celery 可安排周期性的健康状况扫描作业[^2]。 初始化 Flask app 和 Celery worker: ```python from flask import Flask, jsonify, request from celery import Celery app = Flask(__name__) celery_app = Celery('tasks', broker='pyamqp://guest@localhost//') @app.route('/check_network/<string:host>', methods=['GET']) def check_network(host): result = network_check.delay(host) response = {'task_id': result.id} return jsonify(response) @celery_app.task(bind=True) def network_check(self, host): """模拟长时间运行的任务""" import time self.update_state(state="PROGRESS", meta={"current": 0}) time.sleep(5) # 模拟延迟 status = {"status": "UP"} if ping_test(host) else {"status": "DOWN"} return status if __name__ == "__main__": app.run(debug=True) ``` 其中 `ping_test()` 函数可以根据实际情况替换为具体的连通性测试逻辑。 --- #### 4. ### PyEZ 对 Junos 平台的高级操控 PyEZ 是专门为 JUNOS OS 设备开发的一套 SDK 工具包，适用于需要精细调整防火墙规则或是其他复杂场景下的应用场合。 加载模板文件并通过 RPC 方法推送更改至物理机器上： ```python from jnpr.junos import Device from jnpr.junos.utils.config import Config dev_conn = Device(host='router.example.com', user='jcluser', password='password') with dev_conn.open(): cu = Config(dev_conn) template_vars = {...} # 动态填充变量字典 rendered_template = cu.load(template_path='./templates/config_snippet.conf', format='text', **template_vars) diff_output = cu.diff() commit_result = cu.commit(timeout=30) ``` 上述片段体现了怎样依据预定义好的格式迅速部署新设定的能力。 ---